{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Context-Level Document Correction\n", "## Serial Python & 3 Apache SPARK Implementations\n", "***" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "######################\n", "#\n", "# Submission by Gioia Dominedo (Harvard ID: 40966234) for\n", "# CS 205 - Computing Foundations for Computational Science\n", "# \n", "# This is part of a joint project with Kendrick Lo that includes a\n", "# separate component for word-level checking. This script includes \n", "# serial Python code for context-level spell-checking adapted\n", "# from third party algorithms (Symspell and Viterbi algorithms),\n", "# as well as three different SPARK implementations of the same code.\n", "#\n", "# The following were also used as references:\n", "# Peter Norvig, How to Write a Spelling Corrector\n", "# (http://norvig.com/spell-correct.html)\n", "# Peter Norvig, Natural Language Corpus Data: Beautiful Data\n", "# (http://norvig.com/ngrams/ch14.pdf)\n", "#\n", "######################" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "######################\n", "#\n", "# SUMMARY OF CONTEXT-LEVEL CORRECTION LOGIC - VITERBI ALGORITHM\n", "#\n", "# v 1.0 last revised 6 Dec 2015\n", "#\n", "# Each sentence is modeled as a hidden Markov model. Prior\n", "# probabilities (for first words in the sentences) and transition\n", "# probabilities (for all subsequent words) are calculated when\n", "# generating the main dictionary, using the same corpus. Emission\n", "# probabilities are generated on the fly by parameterizing a Poisson \n", "# distribution with the edit distance between words and suggested\n", "# corrections.\n", "#\n", "# The state space of possible corrections for each word is generated\n", "# using logic based on the Symspell spell-checker (see below for more\n", "# detail on Symspell). Valid suggestions must: (a) be 'real' words;\n", "# (b) appear at least 100 times in the corpus used to generate the\n", "# dictionary; (c) be one of the top 10 suggestions, based on frequency\n", "# and edit distance. This simplification ensures that the state space\n", "# remains manageable, though it does reduce the accuracy of the\n", "# suggested corrections.\n", "#\n", "# All probabilities are stored in log-space to avoid underflow. Pre-\n", "# defined minimum values are used for words that are not present in\n", "# the dictionary and/or probability tables.\n", "#\n", "# More detail on the various implementations is included below.\n", "#\n", "######################" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "import re\n", "import math\n", "from scipy.stats import poisson\n", "import time\n", "import itertools" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "import findspark\n", "import os\n", "findspark.init()\n", "import pyspark\n", "sc = pyspark.SparkContext()\n", "sc.setLogLevel('ERROR')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Serial Implementation\n", "***" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Pre-processing" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "######################\n", "#\n", "# PRE-PROCESSING STEPS\n", "#\n", "# The pre-processing steps have been adapted from the dictionary\n", "# creation of the word-level spellchecker, which in turn was based on\n", "# SymSpell, a Symmetric Delete spelling correction algorithm\n", "# developed by Wolf Garbe and originally written in C#. More detail\n", "# on SymSpell is included in the word-level spellcheck documentation.\n", "#\n", "# The main modifications to the word-level spellchecker pre-\n", "# processing stages are to create the additional outputs that are\n", "# required for the context-level checking, and to eliminate redundant\n", "# outputs that are not necessary.\n", "#\n", "# The outputs of the pre-processing stage are:\n", "#\n", "# - dictionary: A dictionary that combines both words present in the\n", "# corpus and other words that are within a given 'delete distance'. \n", "# The format of the dictionary is:\n", "# {word: ([list of words within the given 'delete distance'], \n", "# word count in corpus)}\n", "#\n", "# - start_prob: A dictionary with key, value pairs that correspond to\n", "# (word, probability of the word being the first word in a sentence)\n", "#\n", "# - transition_prob: A dictionary of dictionaries that stores the\n", "# probability of a given word following another. The format of the\n", "# dictionary is:\n", "# {previous word: {word1 : P(word1|prevous word), word2 : \n", "# P(word2|prevous word), ...}}\n", "#\n", "# - default_start_prob: A benchmark probability of a word being at\n", "# the start of a sentence, set to 1 / # of words at the beginning of\n", "# sentences. This ensures that all previously unseen words at the\n", "# beginning of sentences are not corrected unnecessarily.\n", "#\n", "# - default_transition_prob: A benchmark probability of a word being\n", "# seen, given the previous word in the sentence, also set to 1 / # of\n", "# transitions in corpus. This ensures that all previously unseen\n", "# transitions are not corrected unnecessarily.\n", "#\n", "######################\n", "\n", "def get_deletes_list(w, max_edit_distance):\n", " '''\n", " Given a word, derive strings with up to max_edit_distance\n", " characters deleted.\n", " '''\n", " deletes = []\n", " queue = [w]\n", " for d in range(max_edit_distance):\n", " temp_queue = []\n", " for word in queue:\n", " if len(word)>1:\n", " for c in range(len(word)): # character index\n", " word_minus_c = word[:c] + word[c+1:]\n", " if word_minus_c not in deletes:\n", " deletes.append(word_minus_c)\n", " if word_minus_c not in temp_queue:\n", " temp_queue.append(word_minus_c)\n", " queue = temp_queue\n", " \n", " return deletes\n", "\n", "def create_dictionary_entry(w, dictionary, max_edit_distance):\n", " '''\n", " Add a word and its derived deletions to the dictionary.\n", " Dictionary entries are of the form:\n", " ([list of suggested corrections], frequency of word in corpus)\n", " '''\n", "\n", " new_real_word_added = False\n", " \n", " # check if word is already in dictionary\n", " if w in dictionary:\n", " # increment count of word in corpus\n", " dictionary[w] = (dictionary[w][0], dictionary[w][1] + 1)\n", " else:\n", " # create new entry in dictionary\n", " dictionary[w] = ([], 1) \n", " \n", " if dictionary[w][1]==1:\n", " \n", " # first appearance of a word in the corpus\n", " # note: word may already be in dictionary as a derived word\n", " # (e.g. by deleting character from a real word) but the\n", " # word counter frequency is not incremented in those cases\n", " \n", " new_real_word_added = True\n", " deletes = get_deletes_list(w, max_edit_distance)\n", " \n", " for item in deletes:\n", " if item in dictionary:\n", " # add (correct) word to delete's suggested correction\n", " # list if not already there\n", " if item not in dictionary[item][0]:\n", " dictionary[item][0].append(w)\n", " else:\n", " # note: frequency of word in corpus is not incremented\n", " dictionary[item] = ([w], 0) \n", " \n", " return new_real_word_added\n", "\n", "def pre_processing(fname, max_edit_distance=3):\n", " '''\n", " Load a text file and use it to create a dictionary and\n", " to calculate start probabilities and transition probabilities. \n", " '''\n", "\n", " dictionary = dict()\n", " start_prob = dict()\n", " transition_prob = dict()\n", " word_count = 0\n", " transitions = 0\n", " \n", " with open(fname) as file: \n", " \n", " for line in file:\n", " \n", " # process each sentence separately\n", " for sentence in line.replace('?','.').replace('!','.').split('.'):\n", " \n", " # separate by words by non-alphabetical characters\n", " words = re.findall('[a-z]+', sentence.lower()) \n", " \n", " for w, word in enumerate(words):\n", " \n", " # create/update dictionary entry\n", " if create_dictionary_entry(\n", " word, dictionary, max_edit_distance):\n", " word_count += 1\n", " \n", " # update probabilities for Hidden Markov Model\n", " if w == 0:\n", "\n", " # probability of a word being at the\n", " # beginning of a sentence\n", " if word in start_prob:\n", " start_prob[word] += 1\n", " else:\n", " start_prob[word] = 1\n", " else:\n", " \n", " # probability of transitionining from one\n", " # word to another\n", " # dictionary format:\n", " # {previous word: {word1 : P(word1|prevous\n", " # word), word2 : P(word2|prevous word)}}\n", " \n", " # check whether prior word is present\n", " # - create if not\n", " if words[w - 1] not in transition_prob:\n", " transition_prob[words[w - 1]] = dict()\n", " \n", " # check whether current word is present\n", " # - create if not\n", " if word not in transition_prob[words[w - 1]]:\n", " transition_prob[words[w - 1]][word] = 0\n", " \n", " # update value\n", " transition_prob[words[w - 1]][word] += 1\n", " transitions += 1\n", " \n", " # convert counts to log-probabilities, to avoid underflow in\n", " # later calculations (note: natural logarithm, not base-10)\n", "\n", " # also calculate (smalle) default probabilities for words that \n", " # have not already been seen\n", " \n", " # probability of a word being at the beginning of a sentence\n", " total_start_words = float(sum(start_prob.values()))\n", " default_start_prob = math.log(1/total_start_words)\n", " start_prob.update( \n", " {k: math.log(v/total_start_words)\n", " for k, v in start_prob.items()})\n", " \n", " # probability of transitioning from one word to another\n", " default_transition_prob = math.log(1./transitions)\n", " transition_prob.update(\n", " {k: {k1: math.log(float(v1)/sum(v.values()))\n", " for k1, v1 in v.items()} \n", " for k, v in transition_prob.items()})\n", "\n", " # output summary statistics\n", " print 'Total unique words in corpus: %i' % word_count\n", " print 'Total items in dictionary: %i' \\\n", " % len(dictionary)\n", " print ' Edit distance for deletions: %i' % max_edit_distance\n", " print 'Total unique words at the start of a sentence: %i' \\\n", " % len(start_prob)\n", " print 'Total unique word transitions: %i' % len(transition_prob)\n", " \n", " return dictionary, start_prob, default_start_prob, \\\n", " transition_prob, default_transition_prob" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Document correction" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "######################\n", "#\n", "# SPELL-CHECKING - VITERBI ALGORITHM\n", "#\n", "# The below functions are used to read in a text file, break it down\n", "# into individual sentences, and then carry out context-based spell-\n", "# checking on each sentence in turn. In cases where the 'suggested'\n", "# word does not match the actual word in the text, both the original\n", "# and the suggested sentences are printed/outputed to file.\n", "#\n", "# Probabilistic model:\n", "#\n", "# Each sentence is modeled as a hidden Markov model, where the\n", "# hidden states are the words that the user intended to type, and\n", "# the emissions are the words that were actually typed.\n", "#\n", "# For each word in a sentence, we can define:\n", "#\n", "# - emission probabilities: P(observed word|intended word)\n", "#\n", "# - prior probabilities (for first words in sentences only):\n", "# P(being the first word in a sentence)\n", "#\n", "# - transition probabilities (for all subsequent words):\n", "# P(intended word|previous intended word)\n", "#\n", "# Prior and transition probabilities were calculated in the pre-\n", "# processing steps above, using the same corpus as the dictionary.\n", "# \n", "# Emission probabilities are calculated on the fly using a Poisson\n", "# distribution as follows:\n", "# P(observed word|intended word) = PMF of Poisson(k, l), where\n", "# k = edit distance between word typed and word intended, and l=0.01.\n", "# Both the overall approach and the parameter of l=0.01 are based on\n", "# the 2015 lecture notes from AM207 Stochastic Optimization.\n", "# Various parameters for lambda between 0 and 1 were tested, which\n", "# confirmed that 0.01 yields the most accurate word suggestions.\n", "#\n", "# All probabilities are stored in log-space to avoid underflow. Pre-\n", "# defined minimum values (also defined at the pre-processing stage)\n", "# are used for words that are not present in the dictionary and/or\n", "# probability tables.\n", "#\n", "# Algorithm:\n", "#\n", "# The spell-checking itself is carried out using a modified version\n", "# of the Viterbi algorithm, which yields the most likely sequence of\n", "# hidden states, i.e. the most likely sequence of words that form a\n", "# sentence. The main difference to the 'standard' Viterbi algorithm\n", "# is that the state space (i.e. the list of possible corrections) is\n", "# generated (and therefore varies) for each word. This is in contrast\n", "# to the alternative of considering the state space of all possible\n", "# words in the dictionary for every word that is checked, which would\n", "# be intractable for larger dictionaries.\n", "#\n", "# Example:\n", "#\n", "# The algorithm is best illustrated by way of an example.\n", "#\n", "# Suppose that we are checking the sentence 'This is ax test.'\n", "# The emissions for the entire sentence are 'This is ax test.' and\n", "# the hidden states for the entire sentence are 'This is a test.'\n", "#\n", "# As a pre-processing step, we convert everything to lowercase,\n", "# eliminate punctuation, and break the sentence up into a list of\n", "# words: ['this', 'is', 'ax', 'text']\n", "# This list is passed as a parameter to the viterbi function.\n", "#\n", "# The algorithm tackles each word in turn, starting with 'this'.\n", "#\n", "# We first use get_suggestions to obtain a list of all words that\n", "# may have been intended instead of 'this', i.e. all possible hidden\n", "# states (intended words) for the emission (word typed).\n", "#\n", "# get_suggestions returns the 10 most likely corrections:\n", "# - 1 word with an edit distance of 0\n", "# ['this']\n", "# - 3 words with an edit distance of 1\n", "# ['his', 'thus', 'thin']\n", "# - 6 words with an edit distance of 2 \n", "# ['the', 'that', 'is', 'him', 'they', 'their']\n", "# \n", "# These 10 words represent our state space, i.e. possible words that\n", "# may have been intended, and are referred to below as the list of\n", "# possible corrections. They each have an emission probability equal\n", "# to the PMF of Poisson(edit distance, 0.01).\n", "#\n", "# For each word in the list of possible corrections, we calculate:\n", "# P(word starting a sentence) * P(observed 'this'|intended word)\n", "# This is a simple application of Bayes' rule: by normalizing the\n", "# probabilities we obtain P(intended word|oberved 'this') for\n", "# each of the 10 words.\n", "#\n", "# We store the word-probability pairs for future use, and move on to\n", "# the next word. \n", "#\n", "# After the first word, all subsequent words are treated as follows.\n", "#\n", "# The second word in our test sentence is 'is'. Once again, we use\n", "# get_suggestions to obtain a list of all words that may have been\n", "# intended. get_suggestions returns the 10 most likely suggestions:\n", "# - 1 word with an edit distance of 0\n", "# ['is']\n", "# - 9 words with an edit distance of 1\n", "# ['in', 'it', 'his', 'as', 'i', 's', 'if', 'its', 'us']\n", "# These 10 words represent our state space for the second word.\n", "#\n", "# For each word in the current list of possible corrections, we loop\n", "# through all the words in the previous list of possible corrections,\n", "# and calculate:\n", "# probability(previous suggested word) \n", "# * P(current suggested word|previous suggested word)\n", "# * P(typing 'is'|meaning to type current suggested word)\n", "# We determine which previous word maximizes this calculation and\n", "# store that 'path' and probability for each current suggested word.\n", "#\n", "# For example, suppose that we are considering the possibility that\n", "# 'is' was indeed intended to be 'is'. We then calculate: \n", "# probability(previous suggested word)\n", "# * P('is'|previous suggested word) * P('is'|'is')\n", "# for all previous suggested words, and discover that the previous\n", "# suggested word 'this' maximizes the above calculation. We therefore\n", "# store 'this is' as the optimal path for the suggested correction\n", "# 'is' and the above (normalized) probability associated with this\n", "# path.\n", "#\n", "# If the sentence had been only 2 words long, then at this point we\n", "# would return the path that maximizes the most probability for the\n", "# most recent step (word).\n", "#\n", "# As it is not, we repeat the previous steps for 'ax' and 'test',\n", "# and then return the path that is associated with the highest\n", "# probability at the last step.\n", "#\n", "######################\n", "\n", "def dameraulevenshtein(seq1, seq2):\n", " '''\n", " Calculate the Damerau-Levenshtein distance between sequences.\n", "\n", " codesnippet:D0DE4716-B6E6-4161-9219-2903BF8F547F\n", " Conceptually, this is based on a len(seq1) + 1 * len(seq2) + 1\n", " matrix. However, only the current and two previous rows are\n", " needed at once, so we only store those.\n", "\n", " Same code as word-level checking.\n", " '''\n", " \n", " oneago = None\n", " thisrow = range(1, len(seq2) + 1) + [0]\n", " \n", " for x in xrange(len(seq1)):\n", " \n", " twoago, oneago, thisrow = \\\n", " oneago, thisrow, [0] * len(seq2) + [x + 1]\n", " \n", " for y in xrange(len(seq2)):\n", " delcost = oneago[y] + 1\n", " addcost = thisrow[y - 1] + 1\n", " subcost = oneago[y - 1] + (seq1[x] != seq2[y])\n", " thisrow[y] = min(delcost, addcost, subcost)\n", "\n", " if (x > 0 and y > 0 and seq1[x] == seq2[y - 1]\n", " and seq1[x-1] == seq2[y] and seq1[x] != seq2[y]):\n", " thisrow[y] = min(thisrow[y], twoago[y - 2] + 1)\n", " \n", " return thisrow[len(seq2) - 1]\n", "\n", "def get_suggestions(string, dictionary, max_edit_distance, \n", " longest_word_length=20, min_count=100, max_sug=10):\n", " '''\n", " Return list of suggested corrections for potentially incorrectly\n", " spelled word.\n", "\n", " Code based on get_suggestions function from word-level checking,\n", " with the addition of the min_count and max_sug parameters.\n", " - min_count: minimum number of times a word must have appeared\n", " in the dictionary corpus to be considered a valid suggestion\n", " - max_sug: number of suggestions that are returned (ranked by\n", " frequency of appearance in dictionary corpus and edit distance\n", " from word being checked)\n", "\n", " These changes were imposed in order to ensure that the problem\n", " remains tractable when checking very large documents. In practice,\n", " the \"correct\" suggestion is almost always amongst the top ten.\n", "\n", " '''\n", " \n", " if (len(string) - longest_word_length) > max_edit_distance:\n", " # to ensure Viterbi can keep running -- use the word itself\n", " return [(string, 0)]\n", " \n", " suggest_dict = {}\n", " \n", " queue = [string]\n", " q_dictionary = {} # items other than string that we've checked\n", " \n", " while len(queue)>0:\n", " q_item = queue[0] # pop\n", " queue = queue[1:]\n", " \n", " # process queue item\n", " if (q_item in dictionary) and (q_item not in suggest_dict):\n", " if (dictionary[q_item][1]>0):\n", " # word is in dictionary, and is a word from the corpus,\n", " # and not already in suggestion list so add to suggestion\n", " # dictionary, indexed by the word with value (frequency\n", " # in corpus, edit distance)\n", " # note: q_items that are not the input string are shorter\n", " # than input string since only deletes are added (unless\n", " # manual dictionary corrections are added)\n", " assert len(string)>=len(q_item)\n", " suggest_dict[q_item] = \\\n", " (dictionary[q_item][1], len(string) - len(q_item))\n", " \n", " # the suggested corrections for q_item as stored in\n", " # dictionary (whether or not q_item itself is a valid\n", " # word or merely a delete) can be valid corrections\n", " for sc_item in dictionary[q_item][0]:\n", " if (sc_item not in suggest_dict):\n", " \n", " # compute edit distance\n", " # suggested items should always be longer (unless\n", " # manual corrections are added)\n", " assert len(sc_item)>len(q_item)\n", " # q_items that are not input should be shorter\n", " # than original string \n", " # (unless manual corrections added)\n", " assert len(q_item)<=len(string)\n", " if len(q_item)==len(string):\n", " assert q_item==string\n", " item_dist = len(sc_item) - len(q_item)\n", "\n", " # item in suggestions list should not be the same\n", " # as the string itself\n", " assert sc_item!=string \n", " # calculate edit distance using Damerau-\n", " # Levenshtein distance\n", " item_dist = dameraulevenshtein(sc_item, string)\n", " \n", " if item_dist<=max_edit_distance:\n", " # should already be in dictionary if in\n", " # suggestion list\n", " assert sc_item in dictionary \n", " # trim list to contain state space\n", " if (dictionary[q_item][1]>0): \n", " suggest_dict[sc_item] = \\\n", " (dictionary[sc_item][1], item_dist)\n", " \n", " # now generate deletes (e.g. a substring of string or of a\n", " # delete) from the queue item as additional items to check\n", " # -- add to end of queue\n", " assert len(string)>=len(q_item)\n", " if (len(string)-len(q_item))1:\n", " for c in range(len(q_item)): # character index \n", " word_minus_c = q_item[:c] + q_item[c+1:]\n", " if word_minus_c not in q_dictionary:\n", " queue.append(word_minus_c)\n", " # arbitrary value to identify we checked this\n", " q_dictionary[word_minus_c] = None\n", "\n", " # return list of suggestions: (correction, edit distance)\n", " \n", " # only include words that have appeared a minimum number of times\n", " # note: make sure that we do not lose the original word\n", " as_list = [i for i in suggest_dict.items() \n", " if (i[1][0]>min_count or i[0]==string)]\n", " \n", " # only include the most likely suggestions (based on frequency\n", " # and edit distance from original word)\n", " trunc_as_list = sorted(as_list, \n", " key = lambda (term, (freq, dist)): (dist, -freq))[:max_sug]\n", " \n", " if len(trunc_as_list)==0:\n", " # to ensure Viterbi can keep running\n", " # -- use the word itself if no corrections are found\n", " return [(string, 0)]\n", " \n", " else:\n", " # drop the word frequency - not needed beyond this point\n", " return [(i[0], i[1][1]) for i in trunc_as_list]\n", "\n", " '''\n", " Output format:\n", " get_suggestions('file', dictionary)\n", " [('file', 0), ('five', 1), ('fire', 1), ('fine', 1), ('will', 2),\n", " ('time', 2), ('face', 2), ('like', 2), ('life', 2), ('while', 2)]\n", " '''\n", " \n", "def get_emission_prob(edit_dist, poisson_lambda=0.01):\n", " '''\n", " The emission probability, i.e. P(observed word|intended word)\n", " is approximated by a Poisson(k, l) distribution, where \n", " k=edit distance between the observed word and the intended\n", " word and l=0.01.\n", " \n", " Both the overall approach and the parameter of l=0.01 are based on\n", " the 2015 lecture notes from AM207 Stochastic Optimization.\n", " Various parameters for lambda between 0 and 1 were tested, which\n", " confirmed that 0.01 yields the most accurate word suggestions.\n", " '''\n", " \n", " return math.log(poisson.pmf(edit_dist, poisson_lambda))\n", "\n", "######################\n", "#\n", "# Multiple helper functions are used to avoid KeyErrors when\n", "# attempting to access values that are not present in dictionaries,\n", "# in which case the previously specified default value is returned.\n", "#\n", "######################\n", "\n", "def get_start_prob(word, start_prob, default_start_prob):\n", " '''\n", " P(word being at the beginning of a sentence)\n", " '''\n", " try:\n", " return start_prob[word]\n", " except KeyError:\n", " return default_start_prob\n", " \n", "def get_transition_prob(cur_word, prev_word, \n", " transition_prob, default_transition_prob):\n", " '''\n", " P(word|previous word)\n", " '''\n", " try:\n", " return transition_prob[prev_word][cur_word]\n", " except KeyError:\n", " return default_transition_prob\n", "\n", "def get_path_prob(prev_word, prev_path_prob):\n", " '''\n", " P(previous path)\n", " '''\n", " try:\n", " return prev_path_prob[prev_word]\n", " except KeyError:\n", " return math.log(math.exp(min(prev_path_prob.values()))/2.) \n", " \n", "def viterbi(words, dictionary, start_prob, default_start_prob, \n", " transition_prob, default_transition_prob, max_edit_distance):\n", " '''\n", " Determine the most likely (intended) sequence, based on the\n", " observed sequence. Full details in preamble above.\n", " '''\n", "\n", " V = [{}]\n", " path = {}\n", " path_context = []\n", " \n", " # character level correction - used to determine state space\n", " corrections = get_suggestions(words[0], dictionary, max_edit_distance)\n", " \n", " # Initialize base cases (first word in the sentence)\n", " for sug_word in corrections:\n", " \n", " # compute the value for all possible starting states\n", " V[0][sug_word[0]] = math.exp(\n", " get_start_prob(sug_word[0], start_prob, \n", " default_start_prob)\n", " + get_emission_prob(sug_word[1]))\n", " \n", " # remember all the different paths (only one word so far)\n", " path[sug_word[0]] = [sug_word[0]]\n", " \n", " # normalize for numerical stability\n", " path_temp_sum = sum(V[0].values())\n", " V[0].update({k: math.log(v/path_temp_sum) \n", " for k, v in V[0].items()})\n", " \n", " # keep track of previous state space\n", " prev_corrections = [i[0] for i in corrections]\n", " \n", " # return if the sentence only has one word\n", " if len(words) == 1:\n", " path_context = [max(V[0], key=lambda i: V[0][i])]\n", " return path_context\n", "\n", " # run Viterbi for all subsequent words in the sentence\n", " for t in range(1, len(words)):\n", "\n", " V.append({})\n", " new_path = {}\n", " \n", " # character level correction\n", " corrections = get_suggestions(words[t], dictionary, max_edit_distance)\n", " \n", " for sug_word in corrections:\n", " \n", " sug_word_emission_prob = get_emission_prob(sug_word[1])\n", " \n", " # compute the probabilities associated with all previous\n", " # states (paths), only keep the maximum\n", " (prob, word) = max(\n", " (get_path_prob(prev_word, V[t-1]) \n", " + get_transition_prob(sug_word[0], prev_word, \n", " transition_prob, default_transition_prob)\n", " + sug_word_emission_prob, prev_word) \n", " for prev_word in prev_corrections)\n", "\n", " # save the maximum probability for each state\n", " V[t][sug_word[0]] = math.exp(prob)\n", " \n", " # store the full path that results in this probability\n", " new_path[sug_word[0]] = path[word] + [sug_word[0]]\n", " \n", " # normalize for numerical stability\n", " path_temp_sum = sum(V[t].values())\n", " V[t].update({k: math.log(v/path_temp_sum) \n", " for k, v in V[t].items()})\n", " \n", " # keep track of previous state space\n", " prev_corrections = [i[0] for i in corrections]\n", " \n", " # don't need to remember the old paths\n", " path = new_path\n", " \n", " # after all iterations are completed, look up the word with the\n", " # highest probability\n", " (prob, word) = max((V[t][sug_word[0]], sug_word[0]) \n", " for sug_word in corrections)\n", "\n", " # look up the full path associated with this word\n", " path_context = path[word]\n", "\n", " return path_context\n", "\n", "def correct_document_context(fname, dictionary, \n", " start_prob, default_start_prob,\n", " transition_prob, default_transition_prob,\n", " max_edit_distance=3, display_results=False):\n", " \n", " '''\n", " Load a text file and spell-check each sentence using the\n", " dictionary and probability tables that were created in the\n", " pre-processing stage.\n", "\n", " Suggested corrections are either printed to the screen or\n", " saved in a log file, depending on the settings.\n", " '''\n", "\n", " doc_word_count = 0\n", " corrected_word_count = 0\n", " sentence_errors_list = []\n", " total_sentences = 0\n", " \n", " with open(fname) as file:\n", " \n", " for i, line in enumerate(file):\n", " \n", " for sentence in line.replace('?','.').replace('!','.').split('.'):\n", " \n", " # separate by words by non-alphabetical characters\n", " words = re.findall('[a-z]+', sentence.lower()) \n", " doc_word_count += len(words)\n", " \n", " if len(words) > 0:\n", " \n", " # run Viterbi algorithm for each sentence and\n", " # obtain most likely correction (may be the same\n", " # as the original sentence)\n", " suggestion = viterbi(words, dictionary,\n", " start_prob, default_start_prob, \n", " transition_prob, default_transition_prob,\n", " max_edit_distance)\n", "\n", " # display sentences with suggested changes\n", " if words != suggestion:\n", " \n", " # keep track of all potential errors\n", " sentence_errors_list.append([total_sentences, \n", " (words, suggestion)])\n", "\n", " # update count of corrected words\n", " corrected_word_count += \\\n", " sum([words[j]!=suggestion[j] \n", " for j in range(len(words))])\n", " \n", " # used for display purposes\n", " total_sentences += 1\n", " \n", " # print suggested corrections\n", " if display_results:\n", " for sentence in sentence_errors_list:\n", " print 'Sentence %i: %s --> %s' % (sentence[0],\n", " ' '.join(sentence[1][0]), ' '.join(sentence[1][1]))\n", " print '-----'\n", " \n", " # output suggested corrections to file\n", " else:\n", " f = open('spell-log.txt', 'w')\n", " for sentence in sentence_errors_list:\n", " f.write('Sentence %i: %s --> %s\\n' % (sentence[0], \n", " ' '.join(sentence[1][0]), ' '.join(sentence[1][1])))\n", " f.close()\n", " \n", " # display summary statistics\n", " print 'Total words checked: %i' % doc_word_count\n", " print 'Total potential errors found: %i' % corrected_word_count" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Sample performance\n", "The performance of the serial algorithm provides a benchmark against which to compare the subsequent SPARK implementations." ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "dictionary_file = 'testdata/big.txt'\n", "check_file = 'testdata/yelp100reviews.txt'" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "print 'Pre-processing with %s...' % dictionary_file\n", "\n", "start_time = time.time()\n", "\n", "dictionary, start_prob, default_start_prob, \\\n", "transition_prob, default_transition_prob \\\n", "= pre_processing(dictionary_file)\n", "\n", "run_time = time.time() - start_time\n", "\n", "print '-----'\n", "print '%.2f seconds to run' % run_time\n", "print '-----'" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "```\n", "Pre-processing with testdata/big.txt...\n", "Total unique words in corpus: 29157\n", "Total items in dictionary: 2151998\n", " Edit distance for deletions: 3\n", "Total unique words at the start of a sentence: 15356\n", "Total unique word transitions: 27086\n", "-----\n", "32.47 seconds to run\n", "-----\n", "```" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "print 'Spell-checking %s...' % check_file\n", "\n", "start_time = time.time()\n", "\n", "correct_document_context(check_file, dictionary,\n", " start_prob, default_start_prob, \n", " transition_prob, default_transition_prob)\n", "\n", "run_time = time.time() - start_time\n", "\n", "print '-----'\n", "print '%.2f seconds to run' % run_time\n", "print '-----'" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "```\n", "Spell-checking testdata/yelp100reviews.txt...\n", "Total words checked: 12029\n", "Total potential errors found: 1735\n", "-----\n", "303.63 seconds to run\n", "-----\n", "```\n", "Sample output with suggested corrections here." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "
\n", "

NOTE: Three different SPARK implementations were developed and tested. Due to variations in approaches, some of the helper functions vary slightly across implementations but may have the same name. To avoid potential errors, please run the relevant code blocks in each sections before attempting to run the sample code.

\n", "

Please note that runtimes tend to be slightly longer in the iPython notebook when compared to the Python scripts. All runtime results below were derived by running the Python scripts.

\n", "
" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## SPARK Implementation # 1 - Naive Parallelization\n", "***\n", "For this implementation we create an RDD where each element corresponds to a sentence from the document to be checked, and then use a map operation to call the Viterbi function for each sentence.\n", "\n", "This approach takes advantage of parallelization by splitting the sentences among the workers (i.e. increasing the number of workers will improve the runtime), but does not parallelize the algorithm itself.\n", "\n", "Note: all the functions related to the Viterbi algorithm are the same as those used in the serial implementation.\n", "\n", "" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Pre-processing" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "######################\n", "#\n", "# PRE-PROCESSING STEPS\n", "#\n", "# The pre-processing steps have been adapted from the dictionary\n", "# creation of the word-level spellchecker, which in turn was based on\n", "# SymSpell, a Symmetric Delete spelling correction algorithm\n", "# developed by Wolf Garbe and originally written in C#. More detail\n", "# on SymSpell is included in the word-level spellcheck documentation.\n", "#\n", "# The main modifications to the word-level spellchecker pre-\n", "# processing stages are to create the additional outputs that are\n", "# required for the context-level checking, and to eliminate redundant\n", "# outputs that are not necessary.\n", "#\n", "# The outputs of the pre-processing stage are:\n", "#\n", "# - dictionary: A dictionary that combines both words present in the\n", "# corpus and other words that are within a given 'delete distance'. \n", "# The format of the dictionary is:\n", "# {word: ([list of words within the given 'delete distance'], \n", "# word count in corpus)}\n", "#\n", "# - start_prob: A dictionary with key, value pairs that correspond to\n", "# (word, probability of the word being the first word in a sentence)\n", "#\n", "# - transition_prob: A dictionary of dictionaries that stores the\n", "# probability of a given word following another. The format of the\n", "# dictionary is:\n", "# {previous word: {word1 : P(word1|prevous word), word2 : \n", "# P(word2|prevous word), ...}}\n", "#\n", "# - default_start_prob: A benchmark probability of a word being at\n", "# the start of a sentence, set to 1 / # of words at the beginning of\n", "# sentences. This ensures that all previously unseen words at the\n", "# beginning of sentences are not corrected unnecessarily.\n", "#\n", "# - default_transition_prob: A benchmark probability of a word being\n", "# seen, given the previous word in the sentence, also set to 1 / # of\n", "# transitions in corpus. This ensures that all previously unseen\n", "# transitions are not corrected unnecessarily.\n", "#\n", "######################\n", "\n", "def get_deletes_list(w, max_edit_distance):\n", " '''\n", " Given a word, derive strings with up to max_edit_distance\n", " characters deleted. \n", "\n", " The list is generally of the same magnitude as the number of\n", " characters in a word, so it does not make sense to parallelize\n", " this function. Instead, we use Python to create the list.\n", " '''\n", " deletes = []\n", " queue = [w]\n", " for d in range(max_edit_distance):\n", " temp_queue = []\n", " for word in queue:\n", " if len(word)>1:\n", " for c in range(len(word)): # character index\n", " word_minus_c = word[:c] + word[c+1:]\n", " if word_minus_c not in deletes:\n", " deletes.append(word_minus_c)\n", " if word_minus_c not in temp_queue:\n", " temp_queue.append(word_minus_c)\n", " queue = temp_queue\n", " \n", " return deletes\n", "\n", "def get_transitions(sentence):\n", " '''\n", " Helper function: converts a sentence into all two-word pairs.\n", " Output format is a list of tuples.\n", " e.g. 'This is a test' >> ('this', 'is'), ('is', 'a'), ('a', 'test')\n", " ''' \n", " if len(sentence)<2:\n", " return None\n", " else:\n", " return [((sentence[i], sentence[i+1]), 1) \n", " for i in range(len(sentence)-1)]\n", " \n", "def map_transition_prob(vals):\n", " '''\n", " Helper function: calculates conditional probabilities for all word\n", " pairs, i.e. P(word|previous word)\n", " '''\n", " total = float(sum(vals.values()))\n", " return {k: math.log(v/total) for k, v in vals.items()}\n", "\n", "def parallel_create_dictionary(fname, max_edit_distance=3, \n", " num_partitions=6):\n", " '''\n", " Load a text file and use it to create a dictionary and\n", " to calculate start probabilities and transition probabilities. \n", " '''\n", " \n", " # Note: this function makes use of multiple accumulators to keep\n", " # track of the words that are being processed. An alternative \n", " # implementation that wraps accumulators in helper functions was\n", " # also tested, but did not yield any noticeable improvements.\n", "\n", " ############\n", " #\n", " # load file & initial processing\n", " #\n", " ############\n", " \n", " # http://stackoverflow.com/questions/22520932/python-remove-all-non-alphabet-chars-from-string\n", " regex = re.compile('[^a-z ]')\n", "\n", " # load file contents and convert into one long sequence of words\n", " # RDD format: 'line 1', 'line 2', 'line 3', ...\n", " # cache because this RDD is used in multiple operations \n", " make_all_lower = sc.textFile(fname) \\\n", " .map(lambda line: line.lower()) \\\n", " .filter(lambda x: x!='').cache()\n", " \n", " # split into individual sentences and remove other punctuation\n", " # RDD format: [words of sentence 1], [words of sentence 2], ...\n", " # cache because this RDD is used in multiple operations \n", " split_sentence = make_all_lower.flatMap(lambda \n", " line: line.replace('?','.').replace('!','.').split('.')) \\\n", " .map(lambda sentence: regex.sub(' ', sentence)) \\\n", " .map(lambda sentence: sentence.split()) \\\n", " .filter(lambda x: x!=[]).cache()\n", " \n", " ############\n", " #\n", " # generate start probabilities\n", " #\n", " ############\n", " \n", " # extract all words that are at the beginning of sentences\n", " # RDD format: 'word1', 'word2', 'word3', ...\n", " start_words = split_sentence.map(lambda sentence: sentence[0] \n", " if len(sentence)>0 else None) \\\n", " .filter(lambda word: word!=None)\n", " \n", " # add a count to each word\n", " # RDD format: ('word1', 1), ('word2', 1), ('word3', 1), ...\n", " # note: partition here because we are using words as keys for\n", " # the first time - yields a small but consistent improvement in\n", " # runtime (~2-3 sec for big.txt)\n", " # cache because this RDD is used in multiple operations\n", " count_start_words_once = start_words.map(lambda word: (word, 1)) \\\n", " .partitionBy(num_partitions).cache()\n", "\n", " # use accumulator to count the number of start words processed\n", " accum_total_start_words = sc.accumulator(0)\n", " count_start_words_once.foreach(lambda x: accum_total_start_words.add(1))\n", " total_start_words = float(accum_total_start_words.value)\n", " \n", " # reduce into count of unique words at the start of sentences\n", " # RDD format: ('word1', frequency), ('word2', frequency), ...\n", " unique_start_words = count_start_words_once.reduceByKey(lambda a, b: a + b)\n", " \n", " # convert counts to log-probabilities\n", " # RDD format: ('word1', log-prob of word1), \n", " # ('word2', log-prob of word2), ...\n", " start_prob_calc = unique_start_words.mapValues(lambda v: \n", " math.log(v/total_start_words))\n", " \n", " # get default start probabilities (for words not in corpus)\n", " default_start_prob = math.log(1/total_start_words)\n", " \n", " # store start probabilities as a dictionary (i.e. a lookup table)\n", " # note: given the spell-checking algorithm, this cannot be maintained\n", " # as an RDD as it is not possible to map within a map\n", " start_prob = start_prob_calc.collectAsMap()\n", " \n", " ############\n", " #\n", " # generate transition probabilities\n", " #\n", " ############\n", " \n", " # note: various partitioning strategies were attempted for this\n", " # portion of the function, but they failed to yield significant\n", " # improvements in performance.\n", "\n", " # focus on continuous word pairs within the sentence\n", " # e.g. \"this is a test\" -> \"this is\", \"is a\", \"a test\"\n", " # note: as the relevant probability is P(word|previous word)\n", " # the tuples are ordered as (previous word, word)\n", "\n", " # extract all word pairs within a sentence and add a count\n", " # RDD format: (('word1', 'word2'), 1), (('word2', 'word3'), 1), ...\n", " # cache because this RDD is used in multiple operations \n", " other_words = split_sentence.map(lambda sentence: \n", " get_transitions(sentence)) \\\n", " .filter(lambda x: x!=None) \\\n", " .flatMap(lambda x: x).cache()\n", "\n", " # use accumulator to count the number of transitions (word pairs)\n", " accum_total_other_words = sc.accumulator(0)\n", " other_words.foreach(lambda x: accum_total_other_words.add(1))\n", " total_other_words = float(accum_total_other_words.value)\n", " \n", " # reduce into count of unique word pairs\n", " # RDD format: (('word1', 'word2'), frequency), \n", " # (('word2', 'word3'), frequency), ...\n", " unique_other_words = other_words.reduceByKey(lambda a, b: a + b)\n", " \n", " # aggregate by (and change key to) previous word\n", " # RDD format: ('previous word', {'word1': word pair count, \n", " # 'word2': word pair count}}), ...\n", " other_words_collapsed = unique_other_words.map(lambda x: \n", " (x[0][0], (x[0][1], x[1]))) \\\n", " .groupByKey().mapValues(dict)\n", "\n", " # note: the above line of code is the slowest in the function\n", " # (8.6 MB shuffle read and 4.5 MB shuffle write for big.txt)\n", " # An alternative approach that aggregates lists with reduceByKey was\n", " # attempted, but did not yield noticeable improvements in runtime.\n", " \n", " # convert counts to log-probabilities\n", " # RDD format: ('previous word', {'word1': log-prob of pair, \n", " # word2: log-prob of pair}}), ...\n", " transition_prob_calc = other_words_collapsed.mapValues(lambda v: \n", " map_transition_prob(v))\n", "\n", " # get default transition probabilities (for word pairs not in corpus)\n", " default_transition_prob = math.log(1/total_other_words)\n", " \n", " # store transition probabilities as a dictionary (i.e. a lookup table)\n", " # note: given the spell-checking algorithm, this cannot be maintained\n", " # as an RDD as it is not possible to map within a map\n", " transition_prob = transition_prob_calc.collectAsMap()\n", " \n", " ############\n", " #\n", " # generate dictionary\n", " #\n", " ############\n", "\n", " # note: this approach is slightly different from the original SymSpell\n", " # algorithm, but is more appropriate for a SPARK implementation\n", " \n", " # split into individual words (all)\n", " # RDD format: 'word1', 'word2', 'word3', ...\n", " # cache because this RDD is used in multiple operations \n", " all_words = make_all_lower.map(lambda line: regex.sub(' ', line)) \\\n", " .flatMap(lambda line: line.split()).cache()\n", "\n", " # use accumulator to count the number of words processed\n", " accum_words_processed = sc.accumulator(0)\n", " all_words.foreach(lambda x: accum_words_processed.add(1))\n", "\n", " # add a count to each word\n", " # RDD format: ('word1', 1), ('word2', 1), ('word3', 1), ...\n", " count_once = all_words.map(lambda word: (word, 1))\n", "\n", " # reduce into counts of unique words - this is the core corpus dictionary\n", " # (i.e. only words appearing in the file, without 'deletes'))\n", " # RDD format: ('word1', frequency), ('word2', frequency), ...\n", " # cache because this RDD is used in multiple operations \n", " # note: imposing partitioning at this step yields a small \n", " # improvement in runtime (~1 sec for big.txt) by equally\n", " # balancing elements among workers for subsequent operations\n", " unique_words_with_count = count_once.reduceByKey(lambda a, b: a + b, \n", " numPartitions = num_partitions).cache()\n", " \n", " # use accumulator to count the number of unique words\n", " accum_unique_words = sc.accumulator(0)\n", " unique_words_with_count.foreach(lambda x: accum_unique_words.add(1))\n", "\n", " # generate list of \"deletes\" for each word in the corpus\n", " # RDD format: (word1, [deletes for word1]), (word2, [deletes for word2]), ...\n", " generate_deletes = unique_words_with_count.map(lambda (parent, count): \n", " (parent, get_deletes_list(parent, max_edit_distance)))\n", " \n", " # split into all key-value pairs\n", " # RDD format: (word1, delete1), (word1, delete2), ...\n", " expand_deletes = generate_deletes.flatMapValues(lambda x: x)\n", " \n", " # swap word order and add a zero count (because \"deletes\" were not\n", " # present in the dictionary)\n", " swap = expand_deletes.map(lambda (orig, delete): (delete, ([orig], 0)))\n", " \n", " # create a placeholder for each real word\n", " # RDD format: ('word1', ([], frequency)), ('word2', ([], frequency)), ...\n", " corpus = unique_words_with_count.mapValues(lambda count: ([], count))\n", "\n", " # combine main dictionary and \"deletes\" (and eliminate duplicates)\n", " # RDD format: ('word1', ([deletes for word1], frequency)), \n", " # ('word2', ([deletes for word2], frequency)), ...\n", " combine = swap.union(corpus)\n", " \n", " # store dictionary items and deletes as a dictionary (i.e. a lookup table)\n", " # note: given the spell-checking algorithm, this cannot be maintained\n", " # as an RDD as it is not possible to map within a map\n", " # note: use reduceByKeyLocally to avoid an extra shuffle from reduceByKey\n", " dictionary = combine.reduceByKeyLocally(lambda a, b: (a[0]+b[0], a[1]+b[1])) \n", " \n", " # output stats\n", " print 'Total words processed: %i' % accum_words_processed.value\n", " print 'Total unique words in corpus: %i' % accum_unique_words.value \n", " print 'Total items in dictionary (corpus words and deletions): %i' \\\n", " % len(dictionary)\n", " print ' Edit distance for deletions: %i' % max_edit_distance\n", " print 'Total unique words at the start of a sentence: %i' \\\n", " % len(start_prob)\n", " print 'Total unique word transitions: %i' % len(transition_prob)\n", " \n", " return dictionary, start_prob, default_start_prob, \\\n", " transition_prob, default_transition_prob" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Document correction" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "######################\n", "#\n", "# SPELL-CHECKING - VITERBI ALGORITHM\n", "#\n", "# The below functions are used to read in a text file, break it down\n", "# into individual sentences, and then carry out context-based spell-\n", "# checking on each sentence in turn. In cases where the 'suggested'\n", "# word does not match the actual word in the text, both the original\n", "# and the suggested sentences are printed/outputed to file.\n", "#\n", "# Probabilistic model:\n", "#\n", "# Each sentence is modeled as a hidden Markov model, where the\n", "# hidden states are the words that the user intended to type, and\n", "# the emissions are the words that were actually typed.\n", "#\n", "# For each word in a sentence, we can define:\n", "#\n", "# - emission probabilities: P(observed word|intended word)\n", "#\n", "# - prior probabilities (for first words in sentences only):\n", "# P(being the first word in a sentence)\n", "#\n", "# - transition probabilities (for all subsequent words):\n", "# P(intended word|previous intended word)\n", "#\n", "# Prior and transition probabilities were calculated in the pre-\n", "# processing steps above, using the same corpus as the dictionary.\n", "# \n", "# Emission probabilities are calculated on the fly using a Poisson\n", "# distribution as follows:\n", "# P(observed word|intended word) = PMF of Poisson(k, l), where\n", "# k = edit distance between word typed and word intended, and l=0.01.\n", "# Both the overall approach and the parameter of l=0.01 are based on\n", "# the 2015 lecture notes from AM207 Stochastic Optimization.\n", "# Various parameters for lambda between 0 and 1 were tested, which\n", "# confirmed that 0.01 yields the most accurate word suggestions.\n", "#\n", "# All probabilities are stored in log-space to avoid underflow. Pre-\n", "# defined minimum values (also defined at the pre-processing stage)\n", "# are used for words that are not present in the dictionary and/or\n", "# probability tables.\n", "#\n", "# Algorithm:\n", "#\n", "# The spell-checking itself is carried out using a modified version\n", "# of the Viterbi algorithm, which yields the most likely sequence of\n", "# hidden states, i.e. the most likely sequence of words that form a\n", "# sentence. The main difference to the 'standard' Viterbi algorithm\n", "# is that the state space (i.e. the list of possible corrections) is\n", "# generated (and therefore varies) for each word. This is in contrast\n", "# to the alternative of considering the state space of all possible\n", "# words in the dictionary for every word that is checked, which would\n", "# be intractable for larger dictionaries.\n", "#\n", "# Example:\n", "#\n", "# The algorithm is best illustrated by way of an example.\n", "#\n", "# Suppose that we are checking the sentence 'This is ax test.'\n", "# The emissions for the entire sentence are 'This is ax test.' and\n", "# the hidden states for the entire sentence are 'This is a test.'\n", "#\n", "# As a pre-processing step, we convert everything to lowercase,\n", "# eliminate punctuation, and break the sentence up into a list of\n", "# words: ['this', 'is', 'ax', 'text']\n", "# This list is passed as a parameter to the viterbi function.\n", "#\n", "# The algorithm tackles each word in turn, starting with 'this'.\n", "#\n", "# We first use get_suggestions to obtain a list of all words that\n", "# may have been intended instead of 'this', i.e. all possible hidden\n", "# states (intended words) for the emission (word typed).\n", "#\n", "# get_suggestions returns the 10 most likely corrections:\n", "# - 1 word with an edit distance of 0\n", "# ['this']\n", "# - 3 words with an edit distance of 1\n", "# ['his', 'thus', 'thin']\n", "# - 6 words with an edit distance of 2 \n", "# ['the', 'that', 'is', 'him', 'they', 'their']\n", "# \n", "# These 10 words represent our state space, i.e. possible words that\n", "# may have been intended, and are referred to below as the list of\n", "# possible corrections. They each have an emission probability equal\n", "# to the PMF of Poisson(edit distance, 0.01).\n", "#\n", "# For each word in the list of possible corrections, we calculate:\n", "# P(word starting a sentence) * P(observed 'this'|intended word)\n", "# This is a simple application of Bayes' rule: by normalizing the\n", "# probabilities we obtain P(intended word|oberved 'this') for\n", "# each of the 10 words.\n", "#\n", "# We store the word-probability pairs for future use, and move on to\n", "# the next word. \n", "#\n", "# After the first word, all subsequent words are treated as follows.\n", "#\n", "# The second word in our test sentence is 'is'. Once again, we use\n", "# get_suggestions to obtain a list of all words that may have been\n", "# intended. get_suggestions returns the 10 most likely suggestions:\n", "# - 1 word with an edit distance of 0\n", "# ['is']\n", "# - 9 words with an edit distance of 1\n", "# ['in', 'it', 'his', 'as', 'i', 's', 'if', 'its', 'us']\n", "# These 10 words represent our state space for the second word.\n", "#\n", "# For each word in the current list of possible corrections, we loop\n", "# through all the words in the previous list of possible corrections,\n", "# and calculate:\n", "# probability(previous suggested word) \n", "# * P(current suggested word|previous suggested word)\n", "# * P(typing 'is'|meaning to type current suggested word)\n", "# We determine which previous word maximizes this calculation and\n", "# store that 'path' and probability for each current suggested word.\n", "#\n", "# For example, suppose that we are considering the possibility that\n", "# 'is' was indeed intended to be 'is'. We then calculate: \n", "# probability(previous suggested word)\n", "# * P('is'|previous suggested word) * P('is'|'is')\n", "# for all previous suggested words, and discover that the previous\n", "# suggested word 'this' maximizes the above calculation. We therefore\n", "# store 'this is' as the optimal path for the suggested correction\n", "# 'is' and the above (normalized) probability associated with this\n", "# path.\n", "#\n", "# If the sentence had been only 2 words long, then at this point we\n", "# would return the path that maximizes the most probability for the\n", "# most recent step (word).\n", "#\n", "# As it is not, we repeat the previous steps for 'ax' and 'test',\n", "# and then return the path that is associated with the highest\n", "# probability at the last step.\n", "#\n", "######################\n", "\n", "def dameraulevenshtein(seq1, seq2):\n", " '''\n", " Calculate the Damerau-Levenshtein distance between sequences.\n", "\n", " codesnippet:D0DE4716-B6E6-4161-9219-2903BF8F547F\n", " Conceptually, this is based on a len(seq1) + 1 * len(seq2) + 1\n", " matrix. However, only the current and two previous rows are\n", " needed at once, so we only store those.\n", "\n", " Same code as word-level checking.\n", " '''\n", " \n", " oneago = None\n", " thisrow = range(1, len(seq2) + 1) + [0]\n", " \n", " for x in xrange(len(seq1)):\n", " \n", " twoago, oneago, thisrow = \\\n", " oneago, thisrow, [0] * len(seq2) + [x + 1]\n", " \n", " for y in xrange(len(seq2)):\n", " delcost = oneago[y] + 1\n", " addcost = thisrow[y - 1] + 1\n", " subcost = oneago[y - 1] + (seq1[x] != seq2[y])\n", " thisrow[y] = min(delcost, addcost, subcost)\n", "\n", " if (x > 0 and y > 0 and seq1[x] == seq2[y - 1]\n", " and seq1[x-1] == seq2[y] and seq1[x] != seq2[y]):\n", " thisrow[y] = min(thisrow[y], twoago[y - 2] + 1)\n", " \n", " return thisrow[len(seq2) - 1]\n", "\n", "def get_suggestions(string, dictionary, max_edit_distance, \n", " longest_word_length=20, min_count=100, max_sug=10):\n", " '''\n", " Return list of suggested corrections for potentially incorrectly\n", " spelled word.\n", "\n", " Code based on get_suggestions function from word-level checking,\n", " with the addition of the min_count and max_sug parameters.\n", " - min_count: minimum number of times a word must have appeared\n", " in the dictionary corpus to be considered a valid suggestion\n", " - max_sug: number of suggestions that are returned (ranked by\n", " frequency of appearance in dictionary corpus and edit distance\n", " from word being checked)\n", "\n", " These changes were imposed in order to ensure that the problem\n", " remains tractable when checking very large documents. In practice,\n", " the \"correct\" suggestion is almost always amongst the top ten.\n", "\n", " '''\n", " \n", " if (len(string) - longest_word_length) > max_edit_distance:\n", " # to ensure Viterbi can keep running -- use the word itself\n", " return [(string, 0)]\n", " \n", " suggest_dict = {}\n", " \n", " queue = [string]\n", " q_dictionary = {} # items other than string that we've checked\n", " \n", " while len(queue)>0:\n", " q_item = queue[0] # pop\n", " queue = queue[1:]\n", " \n", " # process queue item\n", " if (q_item in dictionary) and (q_item not in suggest_dict):\n", " if (dictionary[q_item][1]>0):\n", " # word is in dictionary, and is a word from the corpus,\n", " # and not already in suggestion list so add to suggestion\n", " # dictionary, indexed by the word with value (frequency\n", " # in corpus, edit distance)\n", " # note: q_items that are not the input string are shorter\n", " # than input string since only deletes are added (unless\n", " # manual dictionary corrections are added)\n", " assert len(string)>=len(q_item)\n", " suggest_dict[q_item] = \\\n", " (dictionary[q_item][1], len(string) - len(q_item))\n", " \n", " # the suggested corrections for q_item as stored in\n", " # dictionary (whether or not q_item itself is a valid\n", " # word or merely a delete) can be valid corrections\n", " for sc_item in dictionary[q_item][0]:\n", " if (sc_item not in suggest_dict):\n", " \n", " # compute edit distance\n", " # suggested items should always be longer (unless\n", " # manual corrections are added)\n", " assert len(sc_item)>len(q_item)\n", " # q_items that are not input should be shorter\n", " # than original string \n", " # (unless manual corrections added)\n", " assert len(q_item)<=len(string)\n", " if len(q_item)==len(string):\n", " assert q_item==string\n", " item_dist = len(sc_item) - len(q_item)\n", "\n", " # item in suggestions list should not be the same\n", " # as the string itself\n", " assert sc_item!=string \n", " # calculate edit distance using Damerau-\n", " # Levenshtein distance\n", " item_dist = dameraulevenshtein(sc_item, string)\n", " \n", " if item_dist<=max_edit_distance:\n", " # should already be in dictionary if in\n", " # suggestion list\n", " assert sc_item in dictionary \n", " # trim list to contain state space\n", " if (dictionary[q_item][1]>0): \n", " suggest_dict[sc_item] = \\\n", " (dictionary[sc_item][1], item_dist)\n", " \n", " # now generate deletes (e.g. a substring of string or of a\n", " # delete) from the queue item as additional items to check\n", " # -- add to end of queue\n", " assert len(string)>=len(q_item)\n", " if (len(string)-len(q_item))1:\n", " for c in range(len(q_item)): # character index \n", " word_minus_c = q_item[:c] + q_item[c+1:]\n", " if word_minus_c not in q_dictionary:\n", " queue.append(word_minus_c)\n", " # arbitrary value to identify we checked this\n", " q_dictionary[word_minus_c] = None\n", "\n", " # return list of suggestions: (correction, edit distance)\n", " \n", " # only include words that have appeared a minimum number of times\n", " # note: make sure that we do not lose the original word\n", " as_list = [i for i in suggest_dict.items() \n", " if (i[1][0]>min_count or i[0]==string)]\n", " \n", " # only include the most likely suggestions (based on frequency\n", " # and edit distance from original word)\n", " trunc_as_list = sorted(as_list, \n", " key = lambda (term, (freq, dist)): (dist, -freq))[:max_sug]\n", " \n", " if len(trunc_as_list)==0:\n", " # to ensure Viterbi can keep running\n", " # -- use the word itself if no corrections are found\n", " return [(string, 0)]\n", " \n", " else:\n", " # drop the word frequency - not needed beyond this point\n", " return [(i[0], i[1][1]) for i in trunc_as_list]\n", "\n", " '''\n", " Output format:\n", " get_suggestions('file', dictionary)\n", " [('file', 0), ('five', 1), ('fire', 1), ('fine', 1), ('will', 2),\n", " ('time', 2), ('face', 2), ('like', 2), ('life', 2), ('while', 2)]\n", " '''\n", " \n", "def get_emission_prob(edit_dist, poisson_lambda=0.01):\n", " '''\n", " The emission probability, i.e. P(observed word|intended word)\n", " is approximated by a Poisson(k, l) distribution, where \n", " k=edit distance between the observed word and the intended\n", " word and l=0.01.\n", " \n", " Both the overall approach and the parameter of l=0.01 are based on\n", " the 2015 lecture notes from AM207 Stochastic Optimization.\n", " Various parameters for lambda between 0 and 1 were tested, which\n", " confirmed that 0.01 yields the most accurate word suggestions.\n", " '''\n", " \n", " return math.log(poisson.pmf(edit_dist, poisson_lambda))\n", "\n", "######################\n", "#\n", "# Multiple helper functions are used to avoid KeyErrors when\n", "# attempting to access values that are not present in dictionaries,\n", "# in which case the previously specified default value is returned.\n", "#\n", "######################\n", "\n", "def get_start_prob(word, start_prob, default_start_prob):\n", " '''\n", " P(word being at the beginning of a sentence)\n", " '''\n", " try:\n", " return start_prob[word]\n", " except KeyError:\n", " return default_start_prob\n", " \n", "def get_transition_prob(cur_word, prev_word, \n", " transition_prob, default_transition_prob):\n", " '''\n", " P(word|previous word)\n", " '''\n", " try:\n", " return transition_prob[prev_word][cur_word]\n", " except KeyError:\n", " return default_transition_prob\n", "\n", "def get_path_prob(prev_word, prev_path_prob):\n", " '''\n", " P(previous path)\n", " '''\n", " try:\n", " return prev_path_prob[prev_word]\n", " except KeyError:\n", " return math.log(math.exp(min(prev_path_prob.values()))/2.) \n", " \n", "def viterbi(words, dictionary, start_prob, default_start_prob, \n", " transition_prob, default_transition_prob, max_edit_distance):\n", " '''\n", " Determine the most likely (intended) sequence, based on the\n", " observed sequence. Full details in preamble above.\n", " '''\n", "\n", " V = [{}]\n", " path = {}\n", " path_context = []\n", " \n", " # character level correction - used to determine state space\n", " corrections = get_suggestions(words[0], dictionary, max_edit_distance)\n", " \n", " # Initialize base cases (first word in the sentence)\n", " for sug_word in corrections:\n", " \n", " # compute the value for all possible starting states\n", " V[0][sug_word[0]] = math.exp(\n", " get_start_prob(sug_word[0], start_prob, \n", " default_start_prob)\n", " + get_emission_prob(sug_word[1]))\n", " \n", " # remember all the different paths (only one word so far)\n", " path[sug_word[0]] = [sug_word[0]]\n", " \n", " # normalize for numerical stability\n", " path_temp_sum = sum(V[0].values())\n", " V[0].update({k: math.log(v/path_temp_sum) \n", " for k, v in V[0].items()})\n", " \n", " # keep track of previous state space\n", " prev_corrections = [i[0] for i in corrections]\n", " \n", " # return if the sentence only has one word\n", " if len(words) == 1:\n", " path_context = [max(V[0], key=lambda i: V[0][i])]\n", " return path_context\n", "\n", " # run Viterbi for all subsequent words in the sentence\n", " for t in range(1, len(words)):\n", "\n", " V.append({})\n", " new_path = {}\n", " \n", " # character level correction\n", " corrections = get_suggestions(words[t], dictionary, max_edit_distance)\n", " \n", " for sug_word in corrections:\n", " \n", " sug_word_emission_prob = get_emission_prob(sug_word[1])\n", " \n", " # compute the probabilities associated with all previous\n", " # states (paths), only keep the maximum\n", " (prob, word) = max(\n", " (get_path_prob(prev_word, V[t-1]) \n", " + get_transition_prob(sug_word[0], prev_word, \n", " transition_prob, default_transition_prob)\n", " + sug_word_emission_prob, prev_word) \n", " for prev_word in prev_corrections)\n", "\n", " # save the maximum probability for each state\n", " V[t][sug_word[0]] = math.exp(prob)\n", " \n", " # store the full path that results in this probability\n", " new_path[sug_word[0]] = path[word] + [sug_word[0]]\n", " \n", " # normalize for numerical stability\n", " path_temp_sum = sum(V[t].values())\n", " V[t].update({k: math.log(v/path_temp_sum) \n", " for k, v in V[t].items()})\n", " \n", " # keep track of previous state space\n", " prev_corrections = [i[0] for i in corrections]\n", " \n", " # don't need to remember the old paths\n", " path = new_path\n", " \n", " # after all iterations are completed, look up the word with the\n", " # highest probability\n", " (prob, word) = max((V[t][sug_word[0]], sug_word[0]) \n", " for sug_word in corrections)\n", "\n", " # look up the full path associated with this word\n", " path_context = path[word]\n", "\n", " return path_context\n", "\n", "def get_count_mismatches(sentences):\n", " '''\n", " Helper function: compares the original sentence with the sentence\n", " that has been suggested by the Viterbi algorithm, and calculates\n", " the number of words that do not match.\n", " '''\n", " orig_sentence, sug_sentence = sentences\n", " count_mismatches = len([(orig_sentence[i], sug_sentence[i]) \n", " for i in range(len(orig_sentence))\n", " if orig_sentence[i]!=sug_sentence[i]])\n", " return count_mismatches, orig_sentence, sug_sentence\n", "\n", "def correct_document_context_parallel_naive(fname, dictionary,\n", " start_prob, default_start_prob,\n", " transition_prob, default_transition_prob,\n", " max_edit_distance=3, num_partitions=6,\n", " display_results=False):\n", " \n", " '''\n", " Load a text file and spell-check each sentence using the\n", " dictionary and probability tables that were created in the\n", " pre-processing stage.\n", "\n", " Suggested corrections are either printed to the screen or\n", " saved in a log file, depending on the settings.\n", " '''\n", "\n", " # note: various partitioning strategies were attempted for this\n", " # function, but they failed to yield significant improvements in\n", " # performance at any file size.\n", "\n", " ############\n", " #\n", " # load file & initial processing\n", " #\n", " ############\n", " \n", " # http://stackoverflow.com/questions/22520932/python-remove-all-non-alphabet-chars-from-string\n", " regex = re.compile('[^a-z ]')\n", "\n", " # broadcast Python dictionaries to workers (from pre-processing)\n", " bc_dictionary = sc.broadcast(dictionary)\n", " bc_start_prob = sc.broadcast(start_prob)\n", " bc_transition_prob = sc.broadcast(transition_prob)\n", " \n", " # load file contents and convert into one long sequence of words\n", " # RDD format: 'line 1', 'line 2', 'line 3', ...\n", " make_all_lower = sc.textFile(fname) \\\n", " .map(lambda line: line.lower()) \\\n", " .filter(lambda x: x!='')\n", " \n", " # split into individual sentences and remove other punctuation\n", " # RDD format: [words of sentence 1], [words of sentence 2], ...\n", " # cache because this RDD is used in multiple operations \n", " split_sentence = make_all_lower.flatMap(lambda \n", " line: line.replace('?','.').replace('!','.').split('.')) \\\n", " .map(lambda sentence: regex.sub(' ', sentence)) \\\n", " .map(lambda sentence: sentence.split()) \\\n", " .filter(lambda x: x!=[]).cache()\n", " \n", " # use accumulator to count the number of words checked\n", " accum_total_words = sc.accumulator(0)\n", " split_sentence.flatMap(lambda x: x) \\\n", " .foreach(lambda x: accum_total_words.add(1))\n", " \n", " # assign a unique id to each sentence\n", " # RDD format: (0, [words of sentence1]), (1, [words of sentence2]), ...\n", " # cache here after completing transformations - results in \n", " # improvements in runtime that scale with file size\n", " # partition as sentence id will remain the key going forward\n", " sentence_id = split_sentence.zipWithIndex().map(\n", " lambda (k, v): (v, k)).partitionBy(num_partitions).cache()\n", "\n", " ############\n", " #\n", " # spell-checking\n", " #\n", " ############\n", "\n", " # use map operation to apply Viterbi algorithm to each sentence\n", " # RDD format: (0, [original sentence1], [corrected sentence1]),\n", " # (1, [original sentence2], [corrected sentence2]), ...\n", " sentence_correction = sentence_id.mapValues(lambda v: (v, \n", " viterbi(v, bc_dictionary.value, bc_start_prob.value, \n", " default_start_prob, bc_transition_prob.value, \n", " default_transition_prob, max_edit_distance)))\n", " ############\n", " #\n", " # output results\n", " #\n", " ############\n", " \n", " # count the number of corrections per sentence and drop any\n", " # sentences without suggested corrections\n", " # RDD format: \n", " # (0, (corrections, [original sentence1], [corrected sentence1])),\n", " # (1, (corrections, [original sentence2], [corrected sentence2])), ...\n", " sentence_errors = sentence_correction.mapValues(lambda v: \n", " (get_count_mismatches(v))). \\\n", " filter(lambda (k, v): v[0]>0)\n", " \n", " # collect all sentences with identified errors (as list)\n", " sentence_errors_list = sentence_errors.collect()\n", " \n", " # count the number of potentially misspelled words\n", " num_errors = sum([s[1][0] for s in sentence_errors_list])\n", " \n", " # print suggested corrections\n", " if display_results:\n", " for sentence in sentence_errors_list:\n", " print 'Sentence %i: %s --> %s' % (sentence[0],\n", " ' '.join(sentence[1][1]), ' '.join(sentence[1][2]))\n", " print '-----'\n", " \n", " # output suggested corrections to file\n", " else:\n", " f = open('spell-log.txt', 'w')\n", " for sentence in sentence_errors_list:\n", " f.write('Sentence %i: %s --> %s\\n' % (sentence[0], \n", " ' '.join(sentence[1][1]), ' '.join(sentence[1][2])))\n", " f.close()\n", " \n", " print '-----'\n", " print 'Total words checked: %i' % accum_total_words.value\n", " print 'Total potential errors found: %i' % num_errors" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Sample performance" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "dictionary_file = 'testdata/big.txt'\n", "check_file = 'testdata/yelp100reviews.txt'" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "print 'Pre-processing with %s...' % dictionary_file\n", "\n", "start_time = time.time()\n", "\n", "dictionary, start_prob, default_start_prob, transition_prob, default_transition_prob = \\\n", " parallel_create_dictionary(dictionary_file)\n", "\n", "run_time = time.time() - start_time\n", "\n", "print '-----'\n", "print '%.2f seconds to run' % run_time\n", "print '-----'" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "```\n", "Pre-processing with testdata/big.txt...\n", "Total words processed: 1105285\n", "Total unique words in corpus: 29157\n", "Total items in dictionary (corpus words and deletions): 2151998\n", " Edit distance for deletions: 3\n", "Total unique words at the start of a sentence: 15356\n", "Total unique word transitions: 27086\n", "-----\n", "54.25 seconds to run\n", "-----\n", "```" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "print 'Spell-checking %s...' % check_file\n", "\n", "start_time = time.time()\n", "\n", "correct_document_context_parallel_naive(check_file, dictionary,\n", " start_prob, default_start_prob, \n", " transition_prob, default_transition_prob)\n", "\n", "run_time = time.time() - start_time\n", "\n", "print '-----'\n", "print '%.2f seconds to run' % run_time\n", "print '-----'" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "```\n", "Spell-checking testdata/yelp100reviews.txt...\n", "-----\n", "Total words checked: 12029\n", "Total potential errors found: 1735\n", "-----\n", "196.14 seconds to run\n", "-----\n", "```\n", "Sample output with suggested corrections here." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Notes on SPARK optimization\n", "\n", "Various optimizations were attempted for this implementation, including modifications to partitioning and caching, as well as slight variations in the operations (e.g. replacing groupByKey with reduceByKey). These changes all failed to yield significant improvements in performance; this is most likely due to the very simple nature of the algorithm, which only has one map operation related to the actual spell-checking. The function correct_document_context_parallel_naive requires very minimal shuffling (~150KB shuffle reads and shuffles writes for 100 Yelp reviews), so there is little room for improvement in that respect.\n", "\n", "It is worth noting the overhead associated with the pre-processing, which is necessary to create the lookup tables that are used by the spell-checker. In particular, creating the transition matrix involves two relatively large shuffle reads and shuffle writes (~8.6MB and ~4.5MB) associated with the reduceByKey and groupByKey operations. The creation of the dictionary and start word probabilities is considerably less expensive (~1MB shuffles for each).\n", "\n", "We found that a relatively small number of partitions (6) performed consistently well across file sizes for both the pre-processing and spell-checking." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Notes on local & AWS performance\n", "\n", "The serial implementation consistently checks approximately 40 words per second. For smaller file sizes, this was clearly superior to the performance of the SPARK code, which was only able to check 4 words per second for 1 Yelp review, and 25 words per second for 10 Yelp reviews. However, the SPARK code began to outperform the serial code when moving to 100 Yelp reviews, achieving 61 words per second, and was able to reach 78 words per second when testing the file with 1,000 Yelp reviews. This file was the largest that could be run locally due to memory constraints.\n", "\n", "Moving to AWS yielded further improvements. The smallest cluster tested (4 executors x 4 cores) was able to achieve 277 words per second on the same 1,000-review file - nearly 4 times the local speed. However, that particular cluster configuration would time out when attempting to run larger file sizes. We also experimented with different numbers of partitions on that cluster, and found that fewer partitions (16) performed slightly better than more partitions (64). Again, this is likely due to the simple nature of the implementation; increasing the number of partitions results in greater complexity without any offsetting gains.\n", "\n", "The next step up was to move to a medium-size cluster (8 executors x 4 cores) and to increase the number of partitions accordingly to 64. This resulted in roughly a doubling in performance, with the same 1,000-review file being processed at 514 words per second. In addition, we were able to run the larger 10,000-review file, with even better performance of 585 words per second.\n", "\n", "The last configuration that we tested was a 16 executor x 4 core cluster. This configuration was clearly most appropriate for files at the larger end of the spectrum: while it only achieved a 1.6x improvement for the 1,000-review file, it ran 2x faster for the 10,000-review file (over 1,100 words per second).\n", "\n", "We had originally planned to test this implementation on even larger clusters with larger files, but were not able to do so due to the limit imposed by Amazon on the total number of instances.\n", "\n", "We used m3.large instances for all of the above configurations, and upgraded the default AWS Python build to version 2.7 for consistency with our local development environment.\n", "\n", "" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## SPARK Implementation # 2 - Approximate Parallelization\n", "***\n", "\n", "This implementation is inspired by the algorithm, but is not completely faithful to it. \n", "\n", "For a given sentence, the Viterbi algorithm steps through each word in turn. At each iteration, it assesses the combination of all previous paths and the word currently under consideration, and only stores the previous path with the highest probability. Consequently, only a subset of the possible combinations of words are retained by the last step.\n", "\n", "In contrast, this implementation considers all the possible sentences that could be created through combinations of all possible corrections for all the words in a sentence.\n", "\n", "e.g. \"This is a test\" -> if each word has 5 possible corrections, then there are 5^4 possible resulting sentences.\n", "\n", "We calculate the probability of each sentence (using the same concepts of start probabilities, emission probabilities, and transition probabilities) and choose the sentence (word combination) with the highest probability.\n", "\n", "There are two main problems with this implementation.\n", "\n", "First, as expected, this approach yields similar corrections to the Viterbi algorithm in many cases, but not always. Despite not being a compeletely faithful representation of the original algorithm, this implementation was considered a way to experiment with a different way of parallelizing the problem.\n", "\n", "The second problem is more severe. Under this implementation, the size of the problem grows exponentially with the size of the text being checked (e.g. an 8-word sentence results in 10^8 possible combinations). This not only makes the approach inefficient, but (more importantly) means that it does not scale to larger problems.\n", "\n", "" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Pre-processing" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "######################\n", "#\n", "# PRE-PROCESSING STEPS\n", "#\n", "# The pre-processing steps have been adapted from the dictionary\n", "# creation of the word-level spellchecker, which in turn was based on\n", "# SymSpell, a Symmetric Delete spelling correction algorithm\n", "# developed by Wolf Garbe and originally written in C#. More detail\n", "# on SymSpell is included in the word-level spellcheck documentation.\n", "#\n", "# The main modifications to the word-level spellchecker pre-\n", "# processing stages are to create the additional outputs that are\n", "# required for the context-level checking, and to eliminate redundant\n", "# outputs that are not necessary.\n", "#\n", "# The outputs of the pre-processing stage are:\n", "#\n", "# - dictionary: A dictionary that combines both words present in the\n", "# corpus and other words that are within a given 'delete distance'. \n", "# The format of the dictionary is:\n", "# {word: ([list of words within the given 'delete distance'], \n", "# word count in corpus)}\n", "#\n", "# - start_prob: A dictionary with key, value pairs that correspond to\n", "# (word, probability of the word being the first word in a sentence)\n", "#\n", "# - transition_prob: A dictionary of dictionaries that stores the\n", "# probability of a given word following another. The format of the\n", "# dictionary is:\n", "# {previous word: {word1 : P(word1|prevous word), word2 : \n", "# P(word2|prevous word), ...}}\n", "#\n", "# - default_start_prob: A benchmark probability of a word being at\n", "# the start of a sentence, set to 1 / # of words at the beginning of\n", "# sentences. This ensures that all previously unseen words at the\n", "# beginning of sentences are not corrected unnecessarily.\n", "#\n", "# - default_transition_prob: A benchmark probability of a word being\n", "# seen, given the previous word in the sentence, also set to 1 / # of\n", "# transitions in corpus. This ensures that all previously unseen\n", "# transitions are not corrected unnecessarily.\n", "#\n", "######################\n", "\n", "def get_deletes_list(w, max_edit_distance):\n", " '''\n", " Given a word, derive strings with up to max_edit_distance\n", " characters deleted. \n", "\n", " The list is generally of the same magnitude as the number of\n", " characters in a word, so it does not make sense to parallelize\n", " this function. Instead, we use Python to create the list.\n", " '''\n", " deletes = []\n", " queue = [w]\n", " for d in range(max_edit_distance):\n", " temp_queue = []\n", " for word in queue:\n", " if len(word)>1:\n", " for c in range(len(word)): # character index\n", " word_minus_c = word[:c] + word[c+1:]\n", " if word_minus_c not in deletes:\n", " deletes.append(word_minus_c)\n", " if word_minus_c not in temp_queue:\n", " temp_queue.append(word_minus_c)\n", " queue = temp_queue\n", " \n", " return deletes\n", "\n", "def get_transitions(sentence):\n", " '''\n", " Helper function: converts a sentence into all two-word pairs.\n", " Output format is a list of tuples.\n", " e.g. 'This is a test' >> ('this', 'is'), ('is', 'a'), ('a', 'test')\n", " ''' \n", " if len(sentence)<2:\n", " return None\n", " else:\n", " return [((sentence[i], sentence[i+1]), 1) \n", " for i in range(len(sentence)-1)]\n", " \n", "def map_transition_prob(vals):\n", " '''\n", " Helper function: calculates conditional probabilities for all word\n", " pairs, i.e. P(word|previous word)\n", " '''\n", " total = float(sum(vals.values()))\n", " return {k: math.log(v/total) for k, v in vals.items()}\n", "\n", "def parallel_create_dictionary(fname, max_edit_distance=3, \n", " num_partitions=6):\n", " '''\n", " Load a text file and use it to create a dictionary and\n", " to calculate start probabilities and transition probabilities. \n", " '''\n", " \n", " # Note: this function makes use of multiple accumulators to keep\n", " # track of the words that are being processed. An alternative \n", " # implementation that wraps accumulators in helper functions was\n", " # also tested, but did not yield any noticeable improvements.\n", "\n", " ############\n", " #\n", " # load file & initial processing\n", " #\n", " ############\n", " \n", " # http://stackoverflow.com/questions/22520932/python-remove-all-non-alphabet-chars-from-string\n", " regex = re.compile('[^a-z ]')\n", "\n", " # load file contents and convert into one long sequence of words\n", " # RDD format: 'line 1', 'line 2', 'line 3', ...\n", " # cache because this RDD is used in multiple operations \n", " make_all_lower = sc.textFile(fname) \\\n", " .map(lambda line: line.lower()) \\\n", " .filter(lambda x: x!='').cache()\n", " \n", " # split into individual sentences and remove other punctuation\n", " # RDD format: [words of sentence 1], [words of sentence 2], ...\n", " # cache because this RDD is used in multiple operations \n", " split_sentence = make_all_lower.flatMap(lambda \n", " line: line.replace('?','.').replace('!','.').split('.')) \\\n", " .map(lambda sentence: regex.sub(' ', sentence)) \\\n", " .map(lambda sentence: sentence.split()) \\\n", " .filter(lambda x: x!=[]).cache()\n", " \n", " ############\n", " #\n", " # generate start probabilities\n", " #\n", " ############\n", " \n", " # extract all words that are at the beginning of sentences\n", " # RDD format: 'word1', 'word2', 'word3', ...\n", " start_words = split_sentence.map(lambda sentence: sentence[0] \n", " if len(sentence)>0 else None) \\\n", " .filter(lambda word: word!=None)\n", " \n", " # add a count to each word\n", " # RDD format: ('word1', 1), ('word2', 1), ('word3', 1), ...\n", " # note: partition here because we are using words as keys for\n", " # the first time - yields a small but consistent improvement in\n", " # runtime (~2-3 sec for big.txt)\n", " # cache because this RDD is used in multiple operations\n", " count_start_words_once = start_words.map(lambda word: (word, 1)) \\\n", " .partitionBy(num_partitions).cache()\n", "\n", " # use accumulator to count the number of start words processed\n", " accum_total_start_words = sc.accumulator(0)\n", " count_start_words_once.foreach(lambda x: accum_total_start_words.add(1))\n", " total_start_words = float(accum_total_start_words.value)\n", " \n", " # reduce into count of unique words at the start of sentences\n", " # RDD format: ('word1', frequency), ('word2', frequency), ...\n", " unique_start_words = count_start_words_once.reduceByKey(lambda a, b: a + b)\n", " \n", " # convert counts to log-probabilities\n", " # RDD format: ('word1', log-prob of word1), \n", " # ('word2', log-prob of word2), ...\n", " start_prob_calc = unique_start_words.mapValues(lambda v: \n", " math.log(v/total_start_words))\n", " \n", " # get default start probabilities (for words not in corpus)\n", " default_start_prob = math.log(1/total_start_words)\n", " \n", " # store start probabilities as a dictionary (i.e. a lookup table)\n", " # note: given the spell-checking algorithm, this cannot be maintained\n", " # as an RDD as it is not possible to map within a map\n", " start_prob = start_prob_calc.collectAsMap()\n", " \n", " ############\n", " #\n", " # generate transition probabilities\n", " #\n", " ############\n", " \n", " # note: various partitioning strategies were attempted for this\n", " # portion of the function, but they failed to yield significant\n", " # improvements in performance.\n", "\n", " # focus on continuous word pairs within the sentence\n", " # e.g. \"this is a test\" -> \"this is\", \"is a\", \"a test\"\n", " # note: as the relevant probability is P(word|previous word)\n", " # the tuples are ordered as (previous word, word)\n", "\n", " # extract all word pairs within a sentence and add a count\n", " # RDD format: (('word1', 'word2'), 1), (('word2', 'word3'), 1), ...\n", " # cache because this RDD is used in multiple operations \n", " other_words = split_sentence.map(lambda sentence: \n", " get_transitions(sentence)) \\\n", " .filter(lambda x: x!=None) \\\n", " .flatMap(lambda x: x).cache()\n", "\n", " # use accumulator to count the number of transitions (word pairs)\n", " accum_total_other_words = sc.accumulator(0)\n", " other_words.foreach(lambda x: accum_total_other_words.add(1))\n", " total_other_words = float(accum_total_other_words.value)\n", " \n", " # reduce into count of unique word pairs\n", " # RDD format: (('word1', 'word2'), frequency), \n", " # (('word2', 'word3'), frequency), ...\n", " unique_other_words = other_words.reduceByKey(lambda a, b: a + b)\n", " \n", " # aggregate by (and change key to) previous word\n", " # RDD format: ('previous word', {'word1': word pair count, \n", " # 'word2': word pair count}}), ...\n", " other_words_collapsed = unique_other_words.map(lambda x: \n", " (x[0][0], (x[0][1], x[1]))) \\\n", " .groupByKey().mapValues(dict)\n", "\n", " # note: the above line of code is the slowest in the function\n", " # (8.6 MB shuffle read and 4.5 MB shuffle write for big.txt)\n", " # An alternative approach that aggregates lists with reduceByKey was\n", " # attempted, but did not yield noticeable improvements in runtime.\n", " \n", " # convert counts to log-probabilities\n", " # RDD format: ('previous word', {'word1': log-prob of pair, \n", " # word2: log-prob of pair}}), ...\n", " transition_prob_calc = other_words_collapsed.mapValues(lambda v: \n", " map_transition_prob(v))\n", "\n", " # get default transition probabilities (for word pairs not in corpus)\n", " default_transition_prob = math.log(1/total_other_words)\n", " \n", " # store transition probabilities as a dictionary (i.e. a lookup table)\n", " # note: given the spell-checking algorithm, this cannot be maintained\n", " # as an RDD as it is not possible to map within a map\n", " transition_prob = transition_prob_calc.collectAsMap()\n", " \n", " ############\n", " #\n", " # generate dictionary\n", " #\n", " ############\n", "\n", " # note: this approach is slightly different from the original SymSpell\n", " # algorithm, but is more appropriate for a SPARK implementation\n", " \n", " # split into individual words (all)\n", " # RDD format: 'word1', 'word2', 'word3', ...\n", " # cache because this RDD is used in multiple operations \n", " all_words = make_all_lower.map(lambda line: regex.sub(' ', line)) \\\n", " .flatMap(lambda line: line.split()).cache()\n", "\n", " # use accumulator to count the number of words processed\n", " accum_words_processed = sc.accumulator(0)\n", " all_words.foreach(lambda x: accum_words_processed.add(1))\n", "\n", " # add a count to each word\n", " # RDD format: ('word1', 1), ('word2', 1), ('word3', 1), ...\n", " count_once = all_words.map(lambda word: (word, 1))\n", "\n", " # reduce into counts of unique words - this is the core corpus dictionary\n", " # (i.e. only words appearing in the file, without 'deletes'))\n", " # RDD format: ('word1', frequency), ('word2', frequency), ...\n", " # cache because this RDD is used in multiple operations \n", " # note: imposing partitioning at this step yields a small \n", " # improvement in runtime (~1 sec for big.txt) by equally\n", " # balancing elements among workers for subsequent operations\n", " unique_words_with_count = count_once.reduceByKey(lambda a, b: a + b, \n", " numPartitions = num_partitions).cache()\n", " \n", " # use accumulator to count the number of unique words\n", " accum_unique_words = sc.accumulator(0)\n", " unique_words_with_count.foreach(lambda x: accum_unique_words.add(1))\n", "\n", " # generate list of \"deletes\" for each word in the corpus\n", " # RDD format: (word1, [deletes for word1]), (word2, [deletes for word2]), ...\n", " generate_deletes = unique_words_with_count.map(lambda (parent, count): \n", " (parent, get_deletes_list(parent, max_edit_distance)))\n", " \n", " # split into all key-value pairs\n", " # RDD format: (word1, delete1), (word1, delete2), ...\n", " expand_deletes = generate_deletes.flatMapValues(lambda x: x)\n", " \n", " # swap word order and add a zero count (because \"deletes\" were not\n", " # present in the dictionary)\n", " swap = expand_deletes.map(lambda (orig, delete): (delete, ([orig], 0)))\n", " \n", " # create a placeholder for each real word\n", " # RDD format: ('word1', ([], frequency)), ('word2', ([], frequency)), ...\n", " corpus = unique_words_with_count.mapValues(lambda count: ([], count))\n", "\n", " # combine main dictionary and \"deletes\" (and eliminate duplicates)\n", " # RDD format: ('word1', ([deletes for word1], frequency)), \n", " # ('word2', ([deletes for word2], frequency)), ...\n", " combine = swap.union(corpus)\n", " \n", " # store dictionary items and deletes as a dictionary (i.e. a lookup table)\n", " # note: given the spell-checking algorithm, this cannot be maintained\n", " # as an RDD as it is not possible to map within a map\n", " # note: use reduceByKeyLocally to avoid an extra shuffle from reduceByKey\n", " dictionary = combine.reduceByKeyLocally(lambda a, b: (a[0]+b[0], a[1]+b[1])) \n", " \n", " # output stats\n", " print 'Total words processed: %i' % accum_words_processed.value\n", " print 'Total unique words in corpus: %i' % accum_unique_words.value \n", " print 'Total items in dictionary (corpus words and deletions): %i' \\\n", " % len(dictionary)\n", " print ' Edit distance for deletions: %i' % max_edit_distance\n", " print 'Total unique words at the start of a sentence: %i' \\\n", " % len(start_prob)\n", " print 'Total unique word transitions: %i' % len(transition_prob)\n", " \n", " return dictionary, start_prob, default_start_prob, \\\n", " transition_prob, default_transition_prob" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Document correction" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "######################\n", "#\n", "# SPELL-CHECKING - (APPROXIMATE) VITERBI ALGORITHM\n", "#\n", "# The below functions are used to read in a text file, break it down\n", "# into individual sentences, and then carry out context-based spell-\n", "# checking on each sentence in turn. In cases where the 'suggested'\n", "# word does not match the actual word in the text, both the original\n", "# and the suggested sentences are printed/outputed to file.\n", "#\n", "# Probabilistic model:\n", "#\n", "# Each sentence is modeled as a hidden Markov model, where the\n", "# hidden states are the words that the user intended to type, and\n", "# the emissions are the words that were actually typed.\n", "#\n", "# For each word in a sentence, we can define:\n", "#\n", "# - emission probabilities: P(observed word|intended word)\n", "#\n", "# - prior probabilities (for first words in sentences only):\n", "# P(being the first word in a sentence)\n", "#\n", "# - transition probabilities (for all subsequent words):\n", "# P(intended word|previous intended word)\n", "#\n", "# Prior and transition probabilities were calculated in the pre-\n", "# processing steps above, using the same corpus as the dictionary.\n", "# \n", "# Emission probabilities are calculated on the fly using a Poisson\n", "# distribution as follows:\n", "# P(observed word|intended word) = PMF of Poisson(k, l), where\n", "# k = edit distance between word typed and word intended, and l=0.01.\n", "# Both the overall approach and the parameter of l=0.01 are based on\n", "# the 2015 lecture notes from AM207 Stochastic Optimization.\n", "# Various parameters for lambda between 0 and 1 were tested, which\n", "# confirmed that 0.01 yields the most accurate word suggestions.\n", "#\n", "# All probabilities are stored in log-space to avoid underflow. Pre-\n", "# defined minimum values (also defined at the pre-processing stage)\n", "# are used for words that are not present in the dictionary and/or\n", "# probability tables.\n", "#\n", "# Algorithm:\n", "#\n", "# The spell-checking itself is carried out using a modified version\n", "# of the Viterbi algorithm, which yields the most likely sequence of\n", "# hidden states, i.e. the most likely sequence of words that form a\n", "# sentence. As with the other implementations, the state space (i.e.\n", "# list of possible corrections) is generated (and therefore varies)\n", "# for each word. This is in contrast to the alternative of considering\n", "# the state space of all possible words in the dictionary for every\n", "# word that is checked, which would be intractable for larger\n", "# dictionaries.\n", "#\n", "# This version also uses a further approximation to the original\n", "# algorithm.\n", "#\n", "# For a given sentence, the Viterbi algorithm steps through each word\n", "# in turn. At each iteration, it assesses the combination of all\n", "# previous paths and the word currently under consideration, and only\n", "# stores the previous path with the highest probability. Consequently,\n", "# only a subset of the possible combinations of words are retained\n", "# by the last step.\n", "#\n", "# In contrast, this implementation considers all the possible\n", "# sentences that could be created through combinations of all possible\n", "# corrections for all the words in a sentence.\n", "# e.g. \"This is a test\" -> if each word has 5 possible corrections,\n", "# then there are 5^4 possible resulting sentences.\n", "# We calculate the probability of each sentence (using the same concepts\n", "# of start probabilities, emission probabilities, and transition\n", "# probabilities) and choose the sentence (word combination) with the\n", "# highest probability.\n", "#\n", "# As expected, we found that this approach yielded similar corrections\n", "# in many cases, but not always. Despite not being a faithful\n", "# representation of the original algorithm, this implementation allowed\n", "# us to experiment with a different way of parallelizing the problem\n", "# of context-level checking.\n", "#\n", "######################\n", "\n", "def dameraulevenshtein(seq1, seq2):\n", " '''\n", " Calculate the Damerau-Levenshtein distance between sequences.\n", "\n", " codesnippet:D0DE4716-B6E6-4161-9219-2903BF8F547F\n", " Conceptually, this is based on a len(seq1) + 1 * len(seq2) + 1\n", " matrix. However, only the current and two previous rows are\n", " needed at once, so we only store those.\n", "\n", " Same code as word-level checking.\n", " '''\n", " \n", " oneago = None\n", " thisrow = range(1, len(seq2) + 1) + [0]\n", " \n", " for x in xrange(len(seq1)):\n", " \n", " twoago, oneago, thisrow = \\\n", " oneago, thisrow, [0] * len(seq2) + [x + 1]\n", " \n", " for y in xrange(len(seq2)):\n", " delcost = oneago[y] + 1\n", " addcost = thisrow[y - 1] + 1\n", " subcost = oneago[y - 1] + (seq1[x] != seq2[y])\n", " thisrow[y] = min(delcost, addcost, subcost)\n", "\n", " if (x > 0 and y > 0 and seq1[x] == seq2[y - 1]\n", " and seq1[x-1] == seq2[y] and seq1[x] != seq2[y]):\n", " thisrow[y] = min(thisrow[y], twoago[y - 2] + 1)\n", " \n", " return thisrow[len(seq2) - 1]\n", "\n", "def get_suggestions(string, dictionary, max_edit_distance, \n", " longest_word_length=20, min_count=100, max_sug=10):\n", " '''\n", " Return list of suggested corrections for potentially incorrectly\n", " spelled word.\n", "\n", " Code based on get_suggestions function from word-level checking,\n", " with the addition of the min_count and max_sug parameters.\n", " - min_count: minimum number of times a word must have appeared\n", " in the dictionary corpus to be considered a valid suggestion\n", " - max_sug: number of suggestions that are returned (ranked by\n", " frequency of appearance in dictionary corpus and edit distance\n", " from word being checked)\n", "\n", " These changes were imposed in order to ensure that the problem\n", " remains tractable when checking very large documents. In practice,\n", " the \"correct\" suggestion is almost always amongst the top ten.\n", "\n", " '''\n", " \n", " if (len(string) - longest_word_length) > max_edit_distance:\n", " # to ensure Viterbi can keep running -- use the word itself\n", " return [(string, 0)]\n", " \n", " suggest_dict = {}\n", " \n", " queue = [string]\n", " q_dictionary = {} # items other than string that we've checked\n", " \n", " while len(queue)>0:\n", " q_item = queue[0] # pop\n", " queue = queue[1:]\n", " \n", " # process queue item\n", " if (q_item in dictionary) and (q_item not in suggest_dict):\n", " if (dictionary[q_item][1]>0):\n", " # word is in dictionary, and is a word from the corpus,\n", " # and not already in suggestion list so add to suggestion\n", " # dictionary, indexed by the word with value (frequency\n", " # in corpus, edit distance)\n", " # note: q_items that are not the input string are shorter\n", " # than input string since only deletes are added (unless\n", " # manual dictionary corrections are added)\n", " assert len(string)>=len(q_item)\n", " suggest_dict[q_item] = \\\n", " (dictionary[q_item][1], len(string) - len(q_item))\n", " \n", " # the suggested corrections for q_item as stored in\n", " # dictionary (whether or not q_item itself is a valid\n", " # word or merely a delete) can be valid corrections\n", " for sc_item in dictionary[q_item][0]:\n", " if (sc_item not in suggest_dict):\n", " \n", " # compute edit distance\n", " # suggested items should always be longer (unless\n", " # manual corrections are added)\n", " assert len(sc_item)>len(q_item)\n", " # q_items that are not input should be shorter\n", " # than original string \n", " # (unless manual corrections added)\n", " assert len(q_item)<=len(string)\n", " if len(q_item)==len(string):\n", " assert q_item==string\n", " item_dist = len(sc_item) - len(q_item)\n", "\n", " # item in suggestions list should not be the same\n", " # as the string itself\n", " assert sc_item!=string \n", " # calculate edit distance using Damerau-\n", " # Levenshtein distance\n", " item_dist = dameraulevenshtein(sc_item, string)\n", " \n", " if item_dist<=max_edit_distance:\n", " # should already be in dictionary if in\n", " # suggestion list\n", " assert sc_item in dictionary \n", " # trim list to contain state space\n", " if (dictionary[q_item][1]>0): \n", " suggest_dict[sc_item] = \\\n", " (dictionary[sc_item][1], item_dist)\n", " \n", " # now generate deletes (e.g. a substring of string or of a\n", " # delete) from the queue item as additional items to check\n", " # -- add to end of queue\n", " assert len(string)>=len(q_item)\n", " if (len(string)-len(q_item))1:\n", " for c in range(len(q_item)): # character index \n", " word_minus_c = q_item[:c] + q_item[c+1:]\n", " if word_minus_c not in q_dictionary:\n", " queue.append(word_minus_c)\n", " # arbitrary value to identify we checked this\n", " q_dictionary[word_minus_c] = None\n", "\n", " # return list of suggestions: (correction, edit distance)\n", " \n", " # only include words that have appeared a minimum number of times\n", " # note: make sure that we do not lose the original word\n", " as_list = [i for i in suggest_dict.items() \n", " if (i[1][0]>min_count or i[0]==string)]\n", " \n", " # only include the most likely suggestions (based on frequency\n", " # and edit distance from original word)\n", " trunc_as_list = sorted(as_list, \n", " key = lambda (term, (freq, dist)): (dist, -freq))[:max_sug]\n", " \n", " if len(trunc_as_list)==0:\n", " # to ensure Viterbi can keep running\n", " # -- use the word itself if no corrections are found\n", " return [(string, 0)]\n", " \n", " else:\n", " # drop the word frequency - not needed beyond this point\n", " return [(i[0], i[1][1]) for i in trunc_as_list]\n", "\n", " '''\n", " Output format:\n", " get_suggestions('file', dictionary)\n", " [('file', 0), ('five', 1), ('fire', 1), ('fine', 1), ('will', 2),\n", " ('time', 2), ('face', 2), ('like', 2), ('life', 2), ('while', 2)]\n", " '''\n", " \n", "def get_emission_prob(edit_dist, poisson_lambda=0.01):\n", " '''\n", " The emission probability, i.e. P(observed word|intended word)\n", " is approximated by a Poisson(k, l) distribution, where \n", " k=edit distance between the observed word and the intended\n", " word and l=0.01.\n", " \n", " Both the overall approach and the parameter of l=0.01 are based on\n", " the 2015 lecture notes from AM207 Stochastic Optimization.\n", " Various parameters for lambda between 0 and 1 were tested, which\n", " confirmed that 0.01 yields the most accurate word suggestions.\n", " '''\n", " \n", " return math.log(poisson.pmf(edit_dist, poisson_lambda))\n", "\n", "######################\n", "# Multiple helper functions are used to avoid KeyErrors when\n", "# attempting to access values that are not present in dictionaries,\n", "# in which case the previously specified default value is returned.\n", "######################\n", "\n", "def get_start_prob(word, start_prob, default_start_prob):\n", " '''\n", " P(word being at the beginning of a sentence)\n", " '''\n", " try:\n", " return start_prob[word]\n", " except KeyError:\n", " return default_start_prob\n", " \n", "def get_transition_prob(cur_word, prev_word, \n", " transition_prob, default_transition_prob):\n", " '''\n", " P(word|previous word)\n", " '''\n", " try:\n", " return transition_prob[prev_word][cur_word]\n", " except KeyError:\n", " return default_transition_prob\n", "\n", "def map_sentence_words(sentence, tmp_dict, max_edit_distance):\n", " '''\n", " Helper function: returns all suggestions for all words\n", " in a sentence.\n", " '''\n", " return [[word, get_suggestions(word, tmp_dict, max_edit_distance)] \n", " for i, word in enumerate(sentence)]\n", "\n", "def split_suggestions_app(sentence):\n", " '''\n", " Helper function: create word-suggestion pairs for each\n", " word in the sentence, and look up the emission probability\n", " of each pair.\n", " '''\n", " result = []\n", " for word in sentence:\n", " result.append([(word[0], s[0], get_emission_prob(s[1])) \n", " for s in word[1]])\n", " return result\n", "\n", "def get_word_combos(sug_lists):\n", " '''\n", " Helper function: returns all possible sentences that can be\n", " created from the list of suggestions for each word in the\n", " original sentence.\n", " e.g. a sentence with 5 words and 10 suggestions per word\n", " will result in 10^6 possible sentences.\n", " '''\n", " return list(itertools.product(*sug_lists))\n", "\n", "def get_combo_prob(combo, tmp_sp, d_sp, tmp_tp, d_tp):\n", " \n", " # first word in sentence\n", " # emission prob * start prob\n", " orig_path = [combo[0][0]]\n", " sug_path = [combo[0][1]]\n", " prob = combo[0][2] + get_start_prob(combo[0][1], tmp_sp, d_sp)\n", " \n", " # subsequent words\n", " for i, w in enumerate(combo[1:]):\n", " orig_path.append(w[0])\n", " sug_path.append(w[1])\n", " prob += w[2] + get_transition_prob(w[1], combo[i-1][1], tmp_tp, d_tp)\n", " \n", " return orig_path, sug_path, prob\n", "\n", "def get_count_mismatches(sentences):\n", " '''\n", " Helper function: compares the original sentence with the sentence\n", " that has been suggested by the Viterbi algorithm, and calculates\n", " the number of words that do not match.\n", " '''\n", " orig_sentence, sug_sentence, prob = sentences\n", " count_mismatches = len([(orig_sentence[i], sug_sentence[i]) \n", " for i in range(len(orig_sentence))\n", " if orig_sentence[i]!=sug_sentence[i]])\n", " return count_mismatches, orig_sentence, sug_sentence\n", "\n", "def correct_document_context_parallel_approximate(fname, dictionary,\n", " start_prob, default_start_prob,\n", " transition_prob, default_transition_prob,\n", " max_edit_distance=3, num_partitions=6,\n", " display_results=False):\n", " \n", " '''\n", " Load a text file and spell-check each sentence using the\n", " dictionary and probability tables that were created in the\n", " pre-processing stage.\n", "\n", " Suggested corrections are either printed to the screen or\n", " saved in a log file, depending on the settings.\n", " '''\n", "\n", " ############\n", " #\n", " # load file & initial processing\n", " #\n", " ############\n", " \n", " # http://stackoverflow.com/questions/22520932/python-remove-all-non-alphabet-chars-from-string\n", " regex = re.compile('[^a-z ]')\n", " \n", " # broadcast Python dictionaries to workers (from pre-processing)\n", " bc_dictionary = sc.broadcast(dictionary)\n", " bc_start_prob = sc.broadcast(start_prob)\n", " bc_transition_prob = sc.broadcast(transition_prob)\n", " \n", " # load file contents and convert into one long sequence of words\n", " # RDD format: 'line 1', 'line 2', 'line 3', ...\n", " make_all_lower = sc.textFile(fname) \\\n", " .map(lambda line: line.lower()) \\\n", " .filter(lambda x: x!='')\n", " \n", " # split into individual sentences and remove other punctuation\n", " # RDD format: [words of sentence 1], [words of sentence 2], ...\n", " # cache because this RDD is used in multiple operations \n", " split_sentence = make_all_lower.flatMap(lambda \n", " line: line.replace('?','.').replace('!','.').split('.')) \\\n", " .map(lambda sentence: regex.sub(' ', sentence)) \\\n", " .map(lambda sentence: sentence.split()) \\\n", " .filter(lambda x: x!=[]).cache()\n", " \n", " # use accumulator to count the number of words checked\n", " accum_total_words = sc.accumulator(0)\n", " split_sentence.flatMap(lambda x: x) \\\n", " .foreach(lambda x: accum_total_words.add(1))\n", " \n", " # assign a unique id to each sentence\n", " # RDD format: (0, [words of sentence1]), (1, [words of sentence2]), ...\n", " # cache here after completing transformations - results in \n", " # improvements in runtime that scale with file size\n", " # partition as sentence id will remain the key going forward\n", " sentence_id = split_sentence.zipWithIndex().map(\n", " lambda (k, v): (v, k)).cache()\n", " \n", " ############\n", " #\n", " # spell-checking\n", " #\n", " ############\n", " \n", " # look up possible suggestions for each word in each sentence\n", " # RDD format:\n", " # (sentence id, [[word, [suggestions]], [word, [suggestions]], ... ]),\n", " # (sentence id, [[word, [suggestions]], [word, [suggestions]], ... ]), ...\n", " sentence_words = sentence_id.mapValues(lambda v: \n", " map_sentence_words(v, bc_dictionary.value, max_edit_distance))\n", " \n", " # look up emission probabilities for each word\n", " # i.e. P(observed word|intended word)\n", " # RDD format: (one list of tuples per word in the sentence)\n", " # (sentence id, [(original word, suggested word, P(original|suggested)),\n", " # (original word, suggested word, P(original|suggested)), ...]), ...\n", " sentence_word_sug = sentence_words.mapValues(lambda v: \n", " split_suggestions_app(v))\n", " \n", " # generate all possible corrected combinations (using Cartesian\n", " # product) - i.e. a sentence with 4 word, each of which have 5 \n", " # possible suggestions, will yield 5^4 possible combinations\n", " # RDD format: (a tuple of tuples for each combination)\n", " # (sentence id, [(original word1, suggested word1, P(original1|suggested1)),\n", " # (original word2, suggested word2, P(original2|suggested2)), ...]), ...\n", " sentence_word_combos = sentence_word_sug.mapValues(lambda v:\n", " get_word_combos(v))\n", " \n", " # flatmap into all possible combinations per sentence\n", " # RDD format: (a tuple of tuples for one combination)\n", " # (sentence id, [(original word1, suggested word1, P(original1|suggested1)),\n", " # (original word2, suggested word2, P(original2|suggested2)), ...]), ...\n", " # partitioning after the flatmap results in a small improvement in runtime\n", " # for the smaller texts on which we were able to run the function\n", " sentence_word_combos_split = sentence_word_combos.flatMapValues(lambda x: x) \\\n", " .partitionBy(num_partitions).cache()\n", " \n", " # calculate the probability of each word combination being the\n", " # intended one, given what was actually typed\n", " # note: the approach does not drop any word combinations, so may\n", " # yield different results to the Viterbi algorithm\n", " # RDD format: \n", " # (sentence id, ([original sentence], [suggested sentence], probability)),\n", " # (sentence id, ([original sentence], [suggested sentence], probability)), ...\n", " sentence_word_combos_prob = sentence_word_combos_split.mapValues(\n", " lambda v: get_combo_prob(v, bc_start_prob.value, default_start_prob, \n", " bc_transition_prob.value, default_transition_prob))\n", " \n", " # identify the word combination with the highest probability for\n", " # each sentence\n", " # (sentence id, ([original sentence], [suggested sentence], probability)),\n", " # (sentence id, ([original sentence], [suggested sentence], probability)), ...\n", " sentence_max_prob = sentence_word_combos_prob.reduceByKey(\n", " lambda a,b: a if a[2] > b[2] else b)\n", "\n", " ############\n", " #\n", " # output results\n", " #\n", " ############\n", "\n", " # count the number of errors per sentence and drop any sentences\n", " # without errors\n", " # RDD format: (sentence id, (# errors, [original sentence], [suggested sentence])),\n", " # (sentence id, (# errors, [original sentence], [suggested sentence])), ...\n", " sentence_errors = sentence_max_prob.mapValues(\n", " lambda v: get_count_mismatches(v)) \\\n", " .filter(lambda (k, v): v[0]>0)\n", "\n", " # collect all sentences with identified errors (as list)\n", " sentence_errors_list = sentence_errors.collect()\n", " \n", " # count the number of potentially misspelled words\n", " num_errors = sum([s[1][0] for s in sentence_errors_list])\n", " \n", " # print suggested corrections\n", " if display_results:\n", " for sentence in sentence_errors_list:\n", " print 'Sentence %i: %s --> %s' % (sentence[0],\n", " ' '.join(sentence[1][1]), ' '.join(sentence[1][2]))\n", " print '-----'\n", " \n", " # output suggested corrections to file\n", " else:\n", " f = open('spell-log.txt', 'w')\n", " for sentence in sentence_errors_list:\n", " f.write('Sentence %i: %s --> %s\\n' % (sentence[0], \n", " ' '.join(sentence[1][1]), ' '.join(sentence[1][2])))\n", " f.close()\n", " \n", " print '-----'\n", " print 'Total words checked: %i' % accum_total_words.value\n", " print 'Total potential errors found: %i' % num_errors" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Sample performance" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "
\n", "WARNING: The memory required by this implementation grows exponentially with the size of the problem. Do not attempt to run this code for larger files.\n", "
" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "dictionary_file = 'testdata/big.txt'\n", "check_file = 'testdata/test.txt'" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "print 'Pre-processing with %s...' % dictionary_file\n", "\n", "start_time = time.time()\n", "\n", "dictionary, start_prob, default_start_prob, transition_prob, default_transition_prob = \\\n", " parallel_create_dictionary(dictionary_file)\n", "\n", "run_time = time.time() - start_time\n", "\n", "print '-----'\n", "print '%.2f seconds to run' % run_time\n", "print '-----'" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "```\n", "Pre-processing with testdata/big.txt...\n", "Total words processed: 1105285\n", "Total unique words in corpus: 29157\n", "Total items in dictionary (corpus words and deletions): 2151998\n", " Edit distance for deletions: 3\n", "Total unique words at the start of a sentence: 15356\n", "Total unique word transitions: 27086\n", "-----\n", "54.25 seconds to run\n", "-----\n", "```" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "print 'Spell-checking %s...' % check_file\n", "\n", "start_time = time.time()\n", "\n", "correct_document_context_parallel_approximate(check_file, dictionary,\n", " start_prob, default_start_prob, \n", " transition_prob, default_transition_prob)\n", "\n", "run_time = time.time() - start_time\n", "\n", "print '-----'\n", "print '%.2f seconds to run' % run_time\n", "print '-----'" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "```\n", "Spell-checking testdata/test.txt...\n", "-----\n", "Total words checked: 27\n", "Total potential errors found: 4\n", "-----\n", "24.99 seconds to run\n", "-----\n", "```\n", "Sample output with suggested corrections here. \n", "Note: the variations in the implementation result in different suggested corrections, compared to the previous or the next implementation. In particular, the running example of \"this is ax test\" is not correctly updated." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Notes on SPARK optimization\n", "This implementation was originally developed and tested on a very small test file (test.txt). It became apparent when moving to larger files that the memory required by this approach grows exponentially with the size of the problem, making it very badly suited for any kind of scaling.\n", "\n", "For this reason, this particular implementation was not taken any further but is included here for completeness." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## SPARK Implementation # 3 - Full Parallelization\n", "***\n", "Unlike previous implementations that relied heavily on helper functions to assess individual sentences, in this version we implement all the steps of the Viterbi algorithm in SPARK.\n", "\n", "The algorithm breaks each sentence to be corrected into its word components, and then loops through each word position exactly as in the original serial algorithm.\n", "\n", "Because different sentences in the document may have different lengths but are all being processed in parallel, we check at each iteration for any sentences that are \"done\" and store their results for later use. Once the maximum sentence length is reached, the results for all of the sentences are processed to extract and display/print suggested corrections.\n", "\n", "This approach takes advantage of parallelization by splitting all the words at a given position among the workers.\n", "\n", "" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Pre-processing" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "######################\n", "#\n", "# PRE-PROCESSING STEPS\n", "#\n", "# The pre-processing steps have been adapted from the dictionary\n", "# creation of the word-level spellchecker, which in turn was based on\n", "# SymSpell, a Symmetric Delete spelling correction algorithm\n", "# developed by Wolf Garbe and originally written in C#. More detail\n", "# on SymSpell is included in the word-level spellcheck documentation.\n", "#\n", "# The main modifications to the word-level spellchecker pre-\n", "# processing stages are to create the additional outputs that are\n", "# required for the context-level checking, and to eliminate redundant\n", "# outputs that are not necessary.\n", "#\n", "# The outputs of the pre-processing stage are:\n", "#\n", "# - dictionary: A dictionary that combines both words present in the\n", "# corpus and other words that are within a given 'delete distance'. \n", "# The format of the dictionary is:\n", "# {word: ([list of words within the given 'delete distance'], \n", "# word count in corpus)}\n", "#\n", "# - start_prob: A dictionary with key, value pairs that correspond to\n", "# (word, probability of the word being the first word in a sentence)\n", "#\n", "# - transition_prob: A dictionary of dictionaries that stores the\n", "# probability of a given word following another. The format of the\n", "# dictionary is:\n", "# {previous word: {word1 : P(word1|prevous word), word2 : \n", "# P(word2|prevous word), ...}}\n", "#\n", "# - default_start_prob: A benchmark probability of a word being at\n", "# the start of a sentence, set to 1 / # of words at the beginning of\n", "# sentences. This ensures that all previously unseen words at the\n", "# beginning of sentences are not corrected unnecessarily.\n", "#\n", "# - default_transition_prob: A benchmark probability of a word being\n", "# seen, given the previous word in the sentence, also set to 1 / # of\n", "# transitions in corpus. This ensures that all previously unseen\n", "# transitions are not corrected unnecessarily.\n", "#\n", "######################\n", "\n", "def get_deletes_list(w, max_edit_distance):\n", " '''\n", " Given a word, derive strings with up to max_edit_distance\n", " characters deleted. \n", "\n", " The list is generally of the same magnitude as the number of\n", " characters in a word, so it does not make sense to parallelize\n", " this function. Instead, we use Python to create the list.\n", " '''\n", " deletes = []\n", " queue = [w]\n", " for d in range(max_edit_distance):\n", " temp_queue = []\n", " for word in queue:\n", " if len(word)>1:\n", " for c in range(len(word)): # character index\n", " word_minus_c = word[:c] + word[c+1:]\n", " if word_minus_c not in deletes:\n", " deletes.append(word_minus_c)\n", " if word_minus_c not in temp_queue:\n", " temp_queue.append(word_minus_c)\n", " queue = temp_queue\n", " \n", " return deletes\n", "\n", "def get_transitions(sentence):\n", " '''\n", " Helper function: converts a sentence into all two-word pairs.\n", " Output format is a list of tuples.\n", " e.g. 'This is a test' >> ('this', 'is'), ('is', 'a'), ('a', 'test')\n", " ''' \n", " if len(sentence)<2:\n", " return None\n", " else:\n", " return [((sentence[i], sentence[i+1]), 1) \n", " for i in range(len(sentence)-1)]\n", " \n", "def map_transition_prob(vals):\n", " '''\n", " Helper function: calculates conditional probabilities for all word\n", " pairs, i.e. P(word|previous word)\n", " '''\n", " total = float(sum(vals.values()))\n", " return {k: math.log(v/total) for k, v in vals.items()}\n", "\n", "def parallel_create_dictionary(fname, max_edit_distance=3, \n", " num_partitions=6):\n", " '''\n", " Load a text file and use it to create a dictionary and\n", " to calculate start probabilities and transition probabilities. \n", " '''\n", " \n", " # Note: this function makes use of multiple accumulators to keep\n", " # track of the words that are being processed. An alternative \n", " # implementation that wraps accumulators in helper functions was\n", " # also tested, but did not yield any noticeable improvements.\n", "\n", " ############\n", " #\n", " # load file & initial processing\n", " #\n", " ############\n", " \n", " # http://stackoverflow.com/questions/22520932/python-remove-all-non-alphabet-chars-from-string\n", " regex = re.compile('[^a-z ]')\n", "\n", " # load file contents and convert into one long sequence of words\n", " # RDD format: 'line 1', 'line 2', 'line 3', ...\n", " # cache because this RDD is used in multiple operations \n", " make_all_lower = sc.textFile(fname) \\\n", " .map(lambda line: line.lower()) \\\n", " .filter(lambda x: x!='').cache()\n", " \n", " # split into individual sentences and remove other punctuation\n", " # RDD format: [words of sentence 1], [words of sentence 2], ...\n", " # cache because this RDD is used in multiple operations \n", " split_sentence = make_all_lower.flatMap(lambda \n", " line: line.replace('?','.').replace('!','.').split('.')) \\\n", " .map(lambda sentence: regex.sub(' ', sentence)) \\\n", " .map(lambda sentence: sentence.split()) \\\n", " .filter(lambda x: x!=[]).cache()\n", " \n", " ############\n", " #\n", " # generate start probabilities\n", " #\n", " ############\n", " \n", " # extract all words that are at the beginning of sentences\n", " # RDD format: 'word1', 'word2', 'word3', ...\n", " start_words = split_sentence.map(lambda sentence: sentence[0] \n", " if len(sentence)>0 else None) \\\n", " .filter(lambda word: word!=None)\n", " \n", " # add a count to each word\n", " # RDD format: ('word1', 1), ('word2', 1), ('word3', 1), ...\n", " # note: partition here because we are using words as keys for\n", " # the first time - yields a small but consistent improvement in\n", " # runtime (~2-3 sec for big.txt)\n", " # cache because this RDD is used in multiple operations\n", " count_start_words_once = start_words.map(lambda word: (word, 1)) \\\n", " .partitionBy(num_partitions).cache()\n", "\n", " # use accumulator to count the number of start words processed\n", " accum_total_start_words = sc.accumulator(0)\n", " count_start_words_once.foreach(lambda x: accum_total_start_words.add(1))\n", " total_start_words = float(accum_total_start_words.value)\n", " \n", " # reduce into count of unique words at the start of sentences\n", " # RDD format: ('word1', frequency), ('word2', frequency), ...\n", " unique_start_words = count_start_words_once.reduceByKey(lambda a, b: a + b)\n", " \n", " # convert counts to log-probabilities\n", " # RDD format: ('word1', log-prob of word1), \n", " # ('word2', log-prob of word2), ...\n", " start_prob_calc = unique_start_words.mapValues(lambda v: \n", " math.log(v/total_start_words))\n", " \n", " # get default start probabilities (for words not in corpus)\n", " default_start_prob = math.log(1/total_start_words)\n", " \n", " # store start probabilities as a dictionary (i.e. a lookup table)\n", " # note: given the spell-checking algorithm, this cannot be maintained\n", " # as an RDD as it is not possible to map within a map\n", " start_prob = start_prob_calc.collectAsMap()\n", " \n", " ############\n", " #\n", " # generate transition probabilities\n", " #\n", " ############\n", " \n", " # note: various partitioning strategies were attempted for this\n", " # portion of the function, but they failed to yield significant\n", " # improvements in performance.\n", "\n", " # focus on continuous word pairs within the sentence\n", " # e.g. \"this is a test\" -> \"this is\", \"is a\", \"a test\"\n", " # note: as the relevant probability is P(word|previous word)\n", " # the tuples are ordered as (previous word, word)\n", "\n", " # extract all word pairs within a sentence and add a count\n", " # RDD format: (('word1', 'word2'), 1), (('word2', 'word3'), 1), ...\n", " # cache because this RDD is used in multiple operations \n", " other_words = split_sentence.map(lambda sentence: \n", " get_transitions(sentence)) \\\n", " .filter(lambda x: x!=None) \\\n", " .flatMap(lambda x: x).cache()\n", "\n", " # use accumulator to count the number of transitions (word pairs)\n", " accum_total_other_words = sc.accumulator(0)\n", " other_words.foreach(lambda x: accum_total_other_words.add(1))\n", " total_other_words = float(accum_total_other_words.value)\n", " \n", " # reduce into count of unique word pairs\n", " # RDD format: (('word1', 'word2'), frequency), \n", " # (('word2', 'word3'), frequency), ...\n", " unique_other_words = other_words.reduceByKey(lambda a, b: a + b)\n", " \n", " # aggregate by (and change key to) previous word\n", " # RDD format: ('previous word', {'word1': word pair count, \n", " # 'word2': word pair count}}), ...\n", " other_words_collapsed = unique_other_words.map(lambda x: \n", " (x[0][0], (x[0][1], x[1]))) \\\n", " .groupByKey().mapValues(dict)\n", "\n", " # note: the above line of code is the slowest in the function\n", " # (8.6 MB shuffle read and 4.5 MB shuffle write for big.txt)\n", " # An alternative approach that aggregates lists with reduceByKey was\n", " # attempted, but did not yield noticeable improvements in runtime.\n", " \n", " # convert counts to log-probabilities\n", " # RDD format: ('previous word', {'word1': log-prob of pair, \n", " # word2: log-prob of pair}}), ...\n", " transition_prob_calc = other_words_collapsed.mapValues(lambda v: \n", " map_transition_prob(v))\n", "\n", " # get default transition probabilities (for word pairs not in corpus)\n", " default_transition_prob = math.log(1/total_other_words)\n", " \n", " # store transition probabilities as a dictionary (i.e. a lookup table)\n", " # note: given the spell-checking algorithm, this cannot be maintained\n", " # as an RDD as it is not possible to map within a map\n", " transition_prob = transition_prob_calc.collectAsMap()\n", " \n", " ############\n", " #\n", " # generate dictionary\n", " #\n", " ############\n", "\n", " # note: this approach is slightly different from the original SymSpell\n", " # algorithm, but is more appropriate for a SPARK implementation\n", " \n", " # split into individual words (all)\n", " # RDD format: 'word1', 'word2', 'word3', ...\n", " # cache because this RDD is used in multiple operations \n", " all_words = make_all_lower.map(lambda line: regex.sub(' ', line)) \\\n", " .flatMap(lambda line: line.split()).cache()\n", "\n", " # use accumulator to count the number of words processed\n", " accum_words_processed = sc.accumulator(0)\n", " all_words.foreach(lambda x: accum_words_processed.add(1))\n", "\n", " # add a count to each word\n", " # RDD format: ('word1', 1), ('word2', 1), ('word3', 1), ...\n", " count_once = all_words.map(lambda word: (word, 1))\n", "\n", " # reduce into counts of unique words - this is the core corpus dictionary\n", " # (i.e. only words appearing in the file, without 'deletes'))\n", " # RDD format: ('word1', frequency), ('word2', frequency), ...\n", " # cache because this RDD is used in multiple operations \n", " # note: imposing partitioning at this step yields a small \n", " # improvement in runtime (~1 sec for big.txt) by equally\n", " # balancing elements among workers for subsequent operations\n", " unique_words_with_count = count_once.reduceByKey(lambda a, b: a + b, \n", " numPartitions = num_partitions).cache()\n", " \n", " # use accumulator to count the number of unique words\n", " accum_unique_words = sc.accumulator(0)\n", " unique_words_with_count.foreach(lambda x: accum_unique_words.add(1))\n", "\n", " # generate list of \"deletes\" for each word in the corpus\n", " # RDD format: (word1, [deletes for word1]), (word2, [deletes for word2]), ...\n", " generate_deletes = unique_words_with_count.map(lambda (parent, count): \n", " (parent, get_deletes_list(parent, max_edit_distance)))\n", " \n", " # split into all key-value pairs\n", " # RDD format: (word1, delete1), (word1, delete2), ...\n", " expand_deletes = generate_deletes.flatMapValues(lambda x: x)\n", " \n", " # swap word order and add a zero count (because \"deletes\" were not\n", " # present in the dictionary)\n", " swap = expand_deletes.map(lambda (orig, delete): (delete, ([orig], 0)))\n", " \n", " # create a placeholder for each real word\n", " # RDD format: ('word1', ([], frequency)), ('word2', ([], frequency)), ...\n", " corpus = unique_words_with_count.mapValues(lambda count: ([], count))\n", "\n", " # combine main dictionary and \"deletes\" (and eliminate duplicates)\n", " # RDD format: ('word1', ([deletes for word1], frequency)), \n", " # ('word2', ([deletes for word2], frequency)), ...\n", " combine = swap.union(corpus)\n", " \n", " # store dictionary items and deletes as a dictionary (i.e. a lookup table)\n", " # note: given the spell-checking algorithm, this cannot be maintained\n", " # as an RDD as it is not possible to map within a map\n", " # note: use reduceByKeyLocally to avoid an extra shuffle from reduceByKey\n", " dictionary = combine.reduceByKeyLocally(lambda a, b: (a[0]+b[0], a[1]+b[1])) \n", " \n", " # output stats\n", " print 'Total words processed: %i' % accum_words_processed.value\n", " print 'Total unique words in corpus: %i' % accum_unique_words.value \n", " print 'Total items in dictionary (corpus words and deletions): %i' \\\n", " % len(dictionary)\n", " print ' Edit distance for deletions: %i' % max_edit_distance\n", " print 'Total unique words at the start of a sentence: %i' \\\n", " % len(start_prob)\n", " print 'Total unique word transitions: %i' % len(transition_prob)\n", " \n", " return dictionary, start_prob, default_start_prob, \\\n", " transition_prob, default_transition_prob" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Document correction" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "######################\n", "#\n", "# SPELL-CHECKING - VITERBI ALGORITHM\n", "#\n", "# The below functions are used to read in a text file, break it down\n", "# into individual sentences, and then carry out context-based spell-\n", "# checking on each sentence in turn. In cases where the 'suggested'\n", "# word does not match the actual word in the text, both the original\n", "# and the suggested sentences are printed/outputed to file.\n", "#\n", "# Probabilistic model:\n", "#\n", "# Each sentence is modeled as a hidden Markov model, where the\n", "# hidden states are the words that the user intended to type, and\n", "# the emissions are the words that were actually typed.\n", "#\n", "# For each word in a sentence, we can define:\n", "#\n", "# - emission probabilities: P(observed word|intended word)\n", "#\n", "# - prior probabilities (for first words in sentences only):\n", "# P(being the first word in a sentence)\n", "#\n", "# - transition probabilities (for all subsequent words):\n", "# P(intended word|previous intended word)\n", "#\n", "# Prior and transition probabilities were calculated in the pre-\n", "# processing steps above, using the same corpus as the dictionary.\n", "# \n", "# Emission probabilities are calculated on the fly using a Poisson\n", "# distribution as follows:\n", "# P(observed word|intended word) = PMF of Poisson(k, l), where\n", "# k = edit distance between word typed and word intended, and l=0.01.\n", "# Both the overall approach and the parameter of l=0.01 are based on\n", "# the 2015 lecture notes from AM207 Stochastic Optimization.\n", "# Various parameters for lambda between 0 and 1 were tested, which\n", "# confirmed that 0.01 yields the most accurate word suggestions.\n", "#\n", "# All probabilities are stored in log-space to avoid underflow. Pre-\n", "# defined minimum values (also defined at the pre-processing stage)\n", "# are used for words that are not present in the dictionary and/or\n", "# probability tables.\n", "#\n", "# Algorithm:\n", "#\n", "# The spell-checking itself is carried out using a modified version\n", "# of the Viterbi algorithm, which yields the most likely sequence of\n", "# hidden states, i.e. the most likely sequence of words that form a\n", "# sentence. The main difference to the 'standard' Viterbi algorithm\n", "# is that the state space (i.e. the list of possible corrections) is\n", "# generated (and therefore varies) for each word. This is in contrast\n", "# to the alternative of considering the state space of all possible\n", "# words in the dictionary for every word that is checked, which would\n", "# be intractable for larger dictionaries.\n", "#\n", "# Example:\n", "#\n", "# The algorithm is best illustrated by way of an example.\n", "#\n", "# Suppose that we are checking the sentence 'This is ax test.'\n", "# The emissions for the entire sentence are 'This is ax test.' and\n", "# the hidden states for the entire sentence are 'This is a test.'\n", "#\n", "# As a pre-processing step, we convert everything to lowercase,\n", "# eliminate punctuation, and break the sentence up into a list of\n", "# words: ['this', 'is', 'ax', 'text']\n", "# This list is passed as a parameter to the viterbi function.\n", "#\n", "# The algorithm tackles each word in turn, starting with 'this'.\n", "#\n", "# We first use get_suggestions to obtain a list of all words that\n", "# may have been intended instead of 'this', i.e. all possible hidden\n", "# states (intended words) for the emission (word typed).\n", "#\n", "# get_suggestions returns the 10 most likely corrections:\n", "# - 1 word with an edit distance of 0\n", "# ['this']\n", "# - 3 words with an edit distance of 1\n", "# ['his', 'thus', 'thin']\n", "# - 6 words with an edit distance of 2 \n", "# ['the', 'that', 'is', 'him', 'they', 'their']\n", "# \n", "# These 10 words represent our state space, i.e. possible words that\n", "# may have been intended, and are referred to below as the list of\n", "# possible corrections. They each have an emission probability equal\n", "# to the PMF of Poisson(edit distance, 0.01).\n", "#\n", "# For each word in the list of possible corrections, we calculate:\n", "# P(word starting a sentence) * P(observed 'this'|intended word)\n", "# This is a simple application of Bayes' rule: by normalizing the\n", "# probabilities we obtain P(intended word|oberved 'this') for\n", "# each of the 10 words.\n", "#\n", "# We store the word-probability pairs for future use, and move on to\n", "# the next word. \n", "#\n", "# After the first word, all subsequent words are treated as follows.\n", "#\n", "# The second word in our test sentence is 'is'. Once again, we use\n", "# get_suggestions to obtain a list of all words that may have been\n", "# intended. get_suggestions returns the 10 most likely suggestions:\n", "# - 1 word with an edit distance of 0\n", "# ['is']\n", "# - 9 words with an edit distance of 1\n", "# ['in', 'it', 'his', 'as', 'i', 's', 'if', 'its', 'us']\n", "# These 10 words represent our state space for the second word.\n", "#\n", "# For each word in the current list of possible corrections, we loop\n", "# through all the words in the previous list of possible corrections,\n", "# and calculate:\n", "# probability(previous suggested word) \n", "# * P(current suggested word|previous suggested word)\n", "# * P(typing 'is'|meaning to type current suggested word)\n", "# We determine which previous word maximizes this calculation and\n", "# store that 'path' and probability for each current suggested word.\n", "#\n", "# For example, suppose that we are considering the possibility that\n", "# 'is' was indeed intended to be 'is'. We then calculate: \n", "# probability(previous suggested word)\n", "# * P('is'|previous suggested word) * P('is'|'is')\n", "# for all previous suggested words, and discover that the previous\n", "# suggested word 'this' maximizes the above calculation. We therefore\n", "# store 'this is' as the optimal path for the suggested correction\n", "# 'is' and the above (normalized) probability associated with this\n", "# path.\n", "#\n", "# If the sentence had been only 2 words long, then at this point we\n", "# would return the path that maximizes the most probability for the\n", "# most recent step (word).\n", "#\n", "# As it is not, we repeat the previous steps for 'ax' and 'test',\n", "# and then return the path that is associated with the highest\n", "# probability at the last step.\n", "#\n", "######################\n", "\n", "def dameraulevenshtein(seq1, seq2):\n", " '''\n", " Calculate the Damerau-Levenshtein distance between sequences.\n", "\n", " codesnippet:D0DE4716-B6E6-4161-9219-2903BF8F547F\n", " Conceptually, this is based on a len(seq1) + 1 * len(seq2) + 1\n", " matrix. However, only the current and two previous rows are\n", " needed at once, so we only store those.\n", "\n", " Same code as word-level checking.\n", " '''\n", " \n", " oneago = None\n", " thisrow = range(1, len(seq2) + 1) + [0]\n", " \n", " for x in xrange(len(seq1)):\n", " \n", " twoago, oneago, thisrow = \\\n", " oneago, thisrow, [0] * len(seq2) + [x + 1]\n", " \n", " for y in xrange(len(seq2)):\n", " delcost = oneago[y] + 1\n", " addcost = thisrow[y - 1] + 1\n", " subcost = oneago[y - 1] + (seq1[x] != seq2[y])\n", " thisrow[y] = min(delcost, addcost, subcost)\n", "\n", " if (x > 0 and y > 0 and seq1[x] == seq2[y - 1]\n", " and seq1[x-1] == seq2[y] and seq1[x] != seq2[y]):\n", " thisrow[y] = min(thisrow[y], twoago[y - 2] + 1)\n", " \n", " return thisrow[len(seq2) - 1]\n", "\n", "def get_suggestions(string, dictionary, max_edit_distance, \n", " longest_word_length=20, min_count=100, max_sug=10):\n", " '''\n", " Return list of suggested corrections for potentially incorrectly\n", " spelled word.\n", "\n", " Code based on get_suggestions function from word-level checking,\n", " with the addition of the min_count and max_sug parameters.\n", " - min_count: minimum number of times a word must have appeared\n", " in the dictionary corpus to be considered a valid suggestion\n", " - max_sug: number of suggestions that are returned (ranked by\n", " frequency of appearance in dictionary corpus and edit distance\n", " from word being checked)\n", "\n", " These changes were imposed in order to ensure that the problem\n", " remains tractable when checking very large documents. In practice,\n", " the \"correct\" suggestion is almost always amongst the top ten.\n", "\n", " '''\n", " \n", " if (len(string) - longest_word_length) > max_edit_distance:\n", " # to ensure Viterbi can keep running -- use the word itself\n", " return [(string, 0)]\n", " \n", " suggest_dict = {}\n", " \n", " queue = [string]\n", " q_dictionary = {} # items other than string that we've checked\n", " \n", " while len(queue)>0:\n", " q_item = queue[0] # pop\n", " queue = queue[1:]\n", " \n", " # process queue item\n", " if (q_item in dictionary) and (q_item not in suggest_dict):\n", " if (dictionary[q_item][1]>0):\n", " # word is in dictionary, and is a word from the corpus,\n", " # and not already in suggestion list so add to suggestion\n", " # dictionary, indexed by the word with value (frequency\n", " # in corpus, edit distance)\n", " # note: q_items that are not the input string are shorter\n", " # than input string since only deletes are added (unless\n", " # manual dictionary corrections are added)\n", " assert len(string)>=len(q_item)\n", " suggest_dict[q_item] = \\\n", " (dictionary[q_item][1], len(string) - len(q_item))\n", " \n", " # the suggested corrections for q_item as stored in\n", " # dictionary (whether or not q_item itself is a valid\n", " # word or merely a delete) can be valid corrections\n", " for sc_item in dictionary[q_item][0]:\n", " if (sc_item not in suggest_dict):\n", " \n", " # compute edit distance\n", " # suggested items should always be longer (unless\n", " # manual corrections are added)\n", " assert len(sc_item)>len(q_item)\n", " # q_items that are not input should be shorter\n", " # than original string \n", " # (unless manual corrections added)\n", " assert len(q_item)<=len(string)\n", " if len(q_item)==len(string):\n", " assert q_item==string\n", " item_dist = len(sc_item) - len(q_item)\n", "\n", " # item in suggestions list should not be the same\n", " # as the string itself\n", " assert sc_item!=string \n", " # calculate edit distance using Damerau-\n", " # Levenshtein distance\n", " item_dist = dameraulevenshtein(sc_item, string)\n", " \n", " if item_dist<=max_edit_distance:\n", " # should already be in dictionary if in\n", " # suggestion list\n", " assert sc_item in dictionary \n", " # trim list to contain state space\n", " if (dictionary[q_item][1]>0): \n", " suggest_dict[sc_item] = \\\n", " (dictionary[sc_item][1], item_dist)\n", " \n", " # now generate deletes (e.g. a substring of string or of a\n", " # delete) from the queue item as additional items to check\n", " # -- add to end of queue\n", " assert len(string)>=len(q_item)\n", " if (len(string)-len(q_item))1:\n", " for c in range(len(q_item)): # character index \n", " word_minus_c = q_item[:c] + q_item[c+1:]\n", " if word_minus_c not in q_dictionary:\n", " queue.append(word_minus_c)\n", " # arbitrary value to identify we checked this\n", " q_dictionary[word_minus_c] = None\n", "\n", " # return list of suggestions: (correction, edit distance)\n", " \n", " # only include words that have appeared a minimum number of times\n", " # note: make sure that we do not lose the original word\n", " as_list = [i for i in suggest_dict.items() \n", " if (i[1][0]>min_count or i[0]==string)]\n", " \n", " # only include the most likely suggestions (based on frequency\n", " # and edit distance from original word)\n", " trunc_as_list = sorted(as_list, \n", " key = lambda (term, (freq, dist)): (dist, -freq))[:max_sug]\n", " \n", " if len(trunc_as_list)==0:\n", " # to ensure Viterbi can keep running\n", " # -- use the word itself if no corrections are found\n", " return [(string, 0)]\n", " \n", " else:\n", " # drop the word frequency - not needed beyond this point\n", " return [(i[0], i[1][1]) for i in trunc_as_list]\n", "\n", " '''\n", " Output format:\n", " get_suggestions('file', dictionary)\n", " [('file', 0), ('five', 1), ('fire', 1), ('fine', 1), ('will', 2),\n", " ('time', 2), ('face', 2), ('like', 2), ('life', 2), ('while', 2)]\n", " '''\n", " \n", "def get_emission_prob(edit_dist, poisson_lambda=0.01):\n", " '''\n", " The emission probability, i.e. P(observed word|intended word)\n", " is approximated by a Poisson(k, l) distribution, where \n", " k=edit distance between the observed word and the intended\n", " word and l=0.01.\n", " \n", " Both the overall approach and the parameter of l=0.01 are based on\n", " the 2015 lecture notes from AM207 Stochastic Optimization.\n", " Various parameters for lambda between 0 and 1 were tested, which\n", " confirmed that 0.01 yields the most accurate word suggestions.\n", " '''\n", " \n", " return math.log(poisson.pmf(edit_dist, poisson_lambda))\n", "\n", "######################\n", "# Multiple helper functions are used to avoid KeyErrors when\n", "# attempting to access values that are not present in dictionaries,\n", "# in which case the previously specified default value is returned.\n", "######################\n", "\n", "def get_start_prob(word, start_prob, default_start_prob):\n", " '''\n", " P(word being at the beginning of a sentence)\n", " '''\n", " try:\n", " return start_prob[word]\n", " except KeyError:\n", " return default_start_prob\n", " \n", "def get_transition_prob(cur_word, prev_word, \n", " transition_prob, default_transition_prob):\n", " '''\n", " P(word|previous word)\n", " '''\n", " try:\n", " return transition_prob[prev_word][cur_word]\n", " except KeyError:\n", " return default_transition_prob \n", "\n", "def get_sentence_word_id(words):\n", " '''\n", " Helper function: numbers each word according to its position\n", " in the sentence.\n", " '''\n", " return [(i, w) for i, w in enumerate(words)]\n", "\n", "def start_word_prob(words, tmp_sp, d_sp):\n", " '''\n", " Helper function: calculates the probability of all word \n", " suggestions being at the beginning of the sentence, based on\n", " the pre-processed start probabilities and the emission model.\n", " i.e. start probability x emission probability\n", " '''\n", " orig_word, sug_words = words\n", " probs = [(w[0], math.exp(\n", " get_start_prob(w[0], tmp_sp, d_sp) \n", " + get_emission_prob(w[1])\n", " )) \n", " for w in sug_words]\n", " sum_probs = sum([p[1] for p in probs])\n", " probs = [([p[0]], math.log(p[1]/sum_probs)) for p in probs]\n", " return probs\n", "\n", "def split_suggestions(sentence):\n", " '''\n", " Helper function: Splits into all the suggestions for a given\n", " word, while retaining the previous path for all elements.\n", " '''\n", " sent_id, (word, word_sug) = sentence\n", " return [[sent_id, (word, w)] for w in word_sug]\n", "\n", "def normalize(probs):\n", " '''\n", " Helper function: normalizes probability so they add to 1.\n", " Note: this is especially necessary given the small\n", " probabilities that apply to this problem.\n", " '''\n", " sum_probs = sum([p[1] for p in probs])\n", " return [(p[0], math.log(p[1]/sum_probs)) for p in probs]\n", "\n", "def get_max_prev_path(words, tmp_tp, d_tp):\n", " '''\n", " Helper function: Calculates the previous path that maximizes\n", " the probability of the current word suggestion.\n", " '''\n", "\n", " # unpack values\n", " cur_word = words[0][0]\n", " cur_sug = words[0][1][0]\n", " cur_sug_ed = words[0][1][1]\n", " prev_sug = words[1]\n", " \n", " # belief + transition probability + emission probability\n", " (prob, word) = max((p[1]\n", " + get_transition_prob(cur_sug, p[0][-1], tmp_tp, d_tp)\n", " + get_emission_prob(cur_sug_ed), p[0])\n", " for p in prev_sug)\n", " \n", " return word + [cur_sug], math.exp(prob)\n", "\n", "def get_max_path(final_paths):\n", " '''\n", " Helper function: at the final step, identifies the full path\n", " (i.e. sentence correction) with the highest probability.\n", " '''\n", " return max((p[1], p[0]) for p in final_paths)[1]\n", "\n", "def get_count_mismatches(sentences):\n", " '''\n", " Helper function: compares the original sentence with the sentence\n", " that has been suggested by the Viterbi algorithm, and calculates\n", " the number of words that do not match.\n", " '''\n", " orig_sentence, sug_sentence = sentences\n", " count_mismatches = len([(orig_sentence[i], sug_sentence[i]) \n", " for i in range(len(orig_sentence))\n", " if orig_sentence[i]!=sug_sentence[i]])\n", " return count_mismatches, orig_sentence, sug_sentence\n", "\n", "def correct_document_context_parallel_full(fname, dictionary,\n", " start_prob, default_start_prob,\n", " transition_prob, default_transition_prob,\n", " max_edit_distance=3, num_partitions=6,\n", " display_results=False):\n", " \n", " '''\n", " Load a text file and spell-check each sentence using the\n", " dictionary and probability tables that were created in the\n", " pre-processing stage.\n", "\n", " Suggested corrections are either printed to the screen or\n", " saved in a log file, depending on the settings.\n", " '''\n", "\n", " ############\n", " #\n", " # load file & initial processing\n", " #\n", " ############\n", " \n", " # http://stackoverflow.com/questions/22520932/python-remove-all-non-alphabet-chars-from-string\n", " regex = re.compile('[^a-z ]')\n", " \n", " # broadcast Python dictionaries to workers (from pre-processing)\n", " bc_dictionary = sc.broadcast(dictionary)\n", " bc_start_prob = sc.broadcast(start_prob)\n", " bc_transition_prob = sc.broadcast(transition_prob)\n", " \n", " # load file contents and convert into one long sequence of words\n", " # RDD format: 'line 1', 'line 2', 'line 3', ...\n", " make_all_lower = sc.textFile(fname) \\\n", " .map(lambda line: line.lower()) \\\n", " .filter(lambda x: x!='')\n", " \n", " # split into individual sentences and remove other punctuation\n", " # RDD format: [words of sentence1], [words of sentence2], ...\n", " # cache because this RDD is used in multiple operations \n", " split_sentence = make_all_lower.flatMap(lambda \n", " line: line.replace('?','.').replace('!','.').split('.')) \\\n", " .map(lambda sentence: regex.sub(' ', sentence)) \\\n", " .map(lambda sentence: sentence.split()) \\\n", " .filter(lambda x: x!=[]).cache()\n", " \n", " # use accumulator to count the number of words checked\n", " accum_total_words = sc.accumulator(0)\n", " split_sentence.flatMap(lambda x: x) \\\n", " .foreach(lambda x: accum_total_words.add(1))\n", " \n", " # assign a unique id to each sentence\n", " # RDD format: (0, [words of sentence1]), (1, [words of sentence2]), ...\n", " # partition and cache here after completing transformations - this\n", " # RDD is used in multiple operations and the sentence id will\n", " # remain the key from this point forward\n", " sentence_id = split_sentence.zipWithIndex().map(\n", " lambda (k, v): (v, k)).partitionBy(num_partitions).cache()\n", " \n", " # count the number of words in each sentence - this is used to\n", " # determine when each sentence is done processing\n", " # RDD format: (0, words in sentence1), (1, words in sentence2), ...\n", " # cache as this RDD is called at every iteration\n", " sentence_word_count = sentence_id.mapValues(lambda v: len(v)).cache()\n", " \n", " ############\n", " #\n", " # spell-checking\n", " #\n", " ############\n", "\n", " # number each word in a sentence, and split into individual words\n", " # RDD format: (sentence1 id, (word1 id, word1)), \n", " # (sentence1 id, (word2 id, word2), ...\n", " sentence_word_id = sentence_id.mapValues(lambda v: get_sentence_word_id(v)) \\\n", " .flatMapValues(lambda x: x)\n", " \n", " # get suggestions for each word\n", " # RDD format: (sentence1 id, (word1 id, word1, [suggestions for word1])), \n", " # (sentence1 id, (word2 id, word2, [suggestions for word2]), ...\n", " # cache as this RDD is called at each iteration\n", " sentence_word_suggestions = sentence_word_id.mapValues(\n", " lambda v: (v[0], v[1], get_suggestions(v[1], bc_dictionary.value,\n", " max_edit_distance))).cache()\n", "\n", " # filter for all the first words in sentences\n", " # RDD format: (sentence id, (0, word, [suggestions for word])), \n", " # (sentence id, (0, word, [suggestions for word]), ...\n", " sentence_word_1 = sentence_word_suggestions.filter(lambda (k, v): v[0]==0) \\\n", " .mapValues(lambda v: (v[1], v[2]))\n", "\n", " # calculate probability for each suggestion\n", " # RDD format: (sentence id, [([word], P(word)), ([word], P(word)), ...]), \n", " # (sentence id, [([word], P(word)), ([word], P(word)), ...]), ...\n", " sentence_path = sentence_word_1.mapValues(lambda v: \n", " start_word_prob(v, bc_start_prob.value, default_start_prob))\n", "\n", " # start loop from second word (zero-indexed)\n", " word_num = 1\n", "\n", " # extract any sentences that have been fully processed\n", " # RDD format: (sentence id, [([path], P(path)), ([path], P(path)), ...]), \n", " # (sentence id, [([path], P(path)), ([path], P(path)), ...]), ...\n", " completed = sentence_word_count.filter(lambda (k, v): v==word_num) \\\n", " .join(sentence_path).mapValues(lambda v: v[1]) \\\n", " .partitionBy(num_partitions).cache()\n", " \n", " # filter for the next words in sentences\n", " # RDD format: (sentence id, (word, [suggestions for word])), \n", " # (sentence id, (word, [suggestions for word]), ...\n", " sentence_word_next = sentence_word_suggestions.filter(lambda \n", " (k,v): v[0]==word_num) \\\n", " .mapValues(lambda v: (v[1], v[2])).cache()\n", " \n", " # check whether there are any words left to process\n", " while not sentence_word_next.isEmpty():\n", "\n", " # split by suggestions, while retaining previous path\n", " # RDD format: (sentence id, (word, (suggested word, edit distance)), \n", " # [previous path]), ...\n", " # use preservesPartitioning to signal that the sentence id\n", " # continues to be the key\n", " sentence_word_next_split = sentence_word_next.flatMap(lambda x: \n", " split_suggestions(x), preservesPartitioning=True)\n", " \n", " # join each suggestion with the previous path\n", " # RDD format:\n", " # (sentence id, ((current word, \n", " # (current word suggestion, edit distance)), \n", " # [(previous path-probability pairs)])), ...\n", " sentence_word_next_path = sentence_word_next_split.join(sentence_path)\n", "\n", " # identify previous path that maximizes the probability \n", " # of each suggested word correction\n", " # RDD format: (sentence id, ([path], path probability)),\n", " # (sentence id, ([path], path probability)), ...\n", " sentence_word_next_path_prob = sentence_word_next_path \\\n", " .mapValues(lambda v: get_max_prev_path(v, \n", " bc_transition_prob.value, default_transition_prob))\n", "\n", " # group all the new paths for each sentence and normalize\n", " # for numerical stability\n", " # RDD format: (sentence id, [([path], P(path)), ([path], P(path)), ...]), \n", " # (sentence id, [([path], P(path)), ([path], P(path)), ...]), ...\n", " # cache as this is used in multiple operations\n", " sentence_path = sentence_word_next_path_prob.groupByKey() \\\n", " .mapValues(lambda v: normalize(v)).cache()\n", "\n", " # move on to next word\n", " word_num += 1\n", " \n", " # extract any sentences that have been fully processed\n", " # RDD format: (sentence id, [([path], P(path)), ([path], P(path)), ...]), \n", " # (sentence id, [([path], P(path)), ([path], P(path)), ...]), ...\n", " # cache as this is carried over to the next iteration\n", " # note: we confirmed that the RDDs being joined/unioned are\n", " # co-partitioned during the development phase\n", " completed = completed \\\n", " .union(sentence_word_count.filter(lambda (k, v): v==word_num) \\\n", " .join(sentence_path) \\\n", " .mapValues(lambda v: v[1])).cache()\n", " \n", " # filter for the next words in sentences\n", " # RDD format: (sentence id, (word, [suggestions for word])), \n", " # (sentence id, (word, [suggestions for word]), ...\n", " # cache as this is carried over to the next iteration\n", " sentence_word_next = sentence_word_suggestions.filter(\n", " lambda (k, v): v[0]==word_num) \\\n", " .mapValues(lambda v: (v[1], v[2])).cache()\n", "\n", " # this is necessary for stability - otherwise too many threads\n", " # are spawned if we collect everything directly below\n", " completed.cache()\n", "\n", " # get most likely path (sentence)\n", " # RDD format: (sentence id, [suggested sentence]),\n", " # (sentence id, [suggested sentence]), ...\n", " sentence_suggestion = completed.mapValues(lambda v: get_max_path(v))\n", "\n", " # checks that RDDs are co-partitioned\n", " # assert sentence_id.partitioner == sentence_suggestion.partitioner\n", "\n", " # join with original path (sentence)\n", " # RDD format: (sentence id, ([original sentence], [suggested sentence])),\n", " # (sentence id, ([original sentence], [suggested sentence])), ...\n", " sentence_max_prob = sentence_id.join(sentence_suggestion)\n", " \n", " ############\n", " #\n", " # output results\n", " #\n", " ############\n", " \n", " # count the number of errors per sentence and drop any sentences\n", " # without errors\n", " # RDD format: (sentence id, (# errors, [original sentence], [suggested sentence])),\n", " # (sentence id, (# errors, [original sentence], [suggested sentence])), ...\n", " sentence_errors = sentence_max_prob.mapValues(\n", " lambda v: get_count_mismatches(v)) \\\n", " .filter(lambda (k, v): v[0]>0)\n", "\n", " # collect all sentences with identified errors (as list)\n", " sentence_errors_list = sentence_errors.collect()\n", "\n", " # count the number of potentially misspelled words\n", " num_errors = sum([s[1][0] for s in sentence_errors_list])\n", " \n", " # print suggested corrections\n", " if display_results:\n", " for sentence in sentence_errors_list:\n", " print 'Sentence %i: %s --> %s' % (sentence[0],\n", " ' '.join(sentence[1][1]), ' '.join(sentence[1][2]))\n", " print '-----'\n", " \n", " # output suggested corrections to file\n", " else:\n", " f = open('spell-log.txt', 'w')\n", " for sentence in sentence_errors_list:\n", " f.write('Sentence %i: %s --> %s\\n' % (sentence[0], \n", " ' '.join(sentence[1][1]), ' '.join(sentence[1][2])))\n", " f.close()\n", " \n", " print '-----'\n", " print 'Total words checked: %i' % accum_total_words.value\n", " print 'Total potential errors found: %i' % num_errors" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Sample performance" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "dictionary_file = 'testdata/big.txt'\n", "check_file = 'testdata/yelp100reviews.txt'" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": true }, "outputs": [], "source": [ "print 'Pre-processing with %s...' % dictionary_file\n", "\n", "start_time = time.time()\n", "\n", "dictionary, start_prob, default_start_prob, transition_prob, default_transition_prob = \\\n", " parallel_create_dictionary(dictionary_file)\n", "\n", "run_time = time.time() - start_time\n", "\n", "print '-----'\n", "print '%.2f seconds to run' % run_time\n", "print '-----'" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "```\n", "Pre-processing with testdata/big.txt...\n", "Total words processed: 1105285\n", "Total unique words in corpus: 29157\n", "Total items in dictionary (corpus words and deletions): 2151998\n", " Edit distance for deletions: 3\n", "Total unique words at the start of a sentence: 15356\n", "Total unique word transitions: 27086\n", "-----\n", "54.25 seconds to run\n", "-----\n", "```" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "collapsed": false }, "outputs": [], "source": [ "print 'Spell-checking %s...' % check_file\n", "\n", "start_time = time.time()\n", "\n", "correct_document_context_parallel_full(check_file, dictionary,\n", " start_prob, default_start_prob, \n", " transition_prob, default_transition_prob)\n", "\n", "run_time = time.time() - start_time\n", "\n", "print '-----'\n", "print '%.2f seconds to run' % run_time\n", "print '-----'" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "```\n", "Spell-checking testdata/yelp100reviews.txt...\n", "-----\n", "Total words checked: 12029\n", "Total potential errors found: 1735\n", "-----\n", "367.10 seconds to run\n", "-----\n", "```\n", "Sample output with suggested corrections here." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Notes on SPARK optimization\n", "\n", "Unlike the previous implementation, the complexity of this approach meant that there was significant scope for optimization. By experimenting with various partitioning and caching strategies, we were able to materially increase the number of words checked per second. The motivation for the changes that were incorporated in the final version are included in-line with the code above.\n", "\n", "One noteworthy point is the memory complexity that is introduced by iterating through each word position for each sentence in parallel (i.e. each RDD element in this iteration is a single word, as opposed to a sentence is the first implementation). When first testing with larger file sizes, we found that this resulted in the code spawning more threads than could be handled by our local configurations. To alleviate this problem, we introduced an additional caching step of the \"completed\" sentences when exiting the loop. This increased the runtime for smaller files sizes, but meant that we were able to run larger file sizes locally.\n", "\n", "Despite all the improvements, the additional complexity of this implementation means that it remains slower overall for all file sizes compared to the naive parallelization approach.\n", "\n", "As with the previous implementation, we found that a relatively small number of partitions (6) performed consistently well across file sizes.\n", "\n", "" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Notes on local & AWS performance\n", "\n", "The serial implementation consistently checks approximately 40 words per second. For smaller file sizes, this was clearly superior to the performance of the SPARK code up to a file size of 100 Yelp reviews (33 words per second in SPARK). However, the SPARK code began to outperform the serial code when moving to larger file sizes, and was able to reach 47 words per second when testing the file with 500 Yelp reviews. This file was the largest that could be run locally due to memory constraints.\n", "\n", "Moving to AWS yielded further improvements, though this implementation did not scale as well as the previous one. For example, the smallest cluster tested (4 executors x 4 cores) was able to achieve 110 words per second on the same 500-review file - just over 2 times the local speed (compared to the 4x speed-up for the first implementation). This improved only marginally to 115 words per second when moving to the 1,000-review file. However, it is interesting to note that increasing the number of partitions to 64 from 16 resulted in further improvements to 130 words per second. Unlike the first implementation, which performed best on this small cluster configuration with 16 partitions, the higher number of RDD elements required for this implementation is better suited to a higher number of partitions, especially when processing larger file sizes.\n", "\n", "Moving to a medium-size cluster (8 executors x 4 cores) resulted in a ~1.7x improvement to 215 words per second, while the large cluster (16 executors x 4 cores) yielded a ~2.2x improvement to 289 words per second (both with respect to the small cluster on the 1,000-review file). This further confirms that this method does not scale to larger cluster sizes as effectively as the first SPARK implementation.\n", "\n", "Once again, we used m3.large instances for all of the above configurations, and upgraded the default AWS Python build to version 2.7 for consistency with the local development environment.\n", "\n", "" ] } ], "metadata": { "kernelspec": { "display_name": "Python 2", "language": "python", "name": "python2" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 2 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython2", "version": "2.7.10" } }, "nbformat": 4, "nbformat_minor": 0 }